D-Net

package eu.dnetlib.data.mapreduce.hbase.dataexport;

import com.fasterxml.jackson.annotation.JsonInclude;

import com.fasterxml.jackson.databind.ObjectMapper;

import com.google.gson.Gson;

import com.google.protobuf.InvalidProtocolBufferException;

import eu.dnetlib.data.mapreduce.util.DedupUtils;

import eu.dnetlib.data.mapreduce.util.OafRowKeyDecoder;

import eu.dnetlib.data.mapreduce.util.UpdateMerger;

import eu.dnetlib.data.proto.OafProtos;

import eu.dnetlib.data.proto.TypeProtos;

import eu.dnetlib.dhp.schema.oaf.Oaf;

import eu.dnetlib.dhp.schema.oaf.OafEntity;

import eu.dnetlib.dhp.schema.oaf.Relation;

import org.apache.commons.lang.StringUtils;

import org.apache.hadoop.hbase.client.Result;

import org.apache.hadoop.hbase.io.ImmutableBytesWritable;

import org.apache.hadoop.hbase.mapreduce.TableMapper;

import org.apache.hadoop.hbase.util.Bytes;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;

import java.io.IOException;

import java.util.Map;

import java.util.Map.Entry;

import java.util.NavigableMap;

import java.util.stream.Collectors;

/**

 * Exports Oaf objects as their json serialization.

 *

 * @author claudio

 *

 */

public class ExportInformationSpaceMapper2DHP extends TableMapper<Text, Text> {

	private Text keyOut;

	private Text valueOut;

	private MultipleOutputs multipleOutputs;

	private ObjectMapper objectMapper;

	@Override

	protected void setup(final Context context) throws IOException, InterruptedException {

		super.setup(context);

		keyOut = new Text();

		valueOut = new Text();

		multipleOutputs = new MultipleOutputs(context);

		objectMapper = new ObjectMapper()

				.setSerializationInclusion(JsonInclude.Include.NON_NULL);

	}

	@Override

	protected void cleanup(Context context) throws IOException, InterruptedException {

		multipleOutputs.close();

	}

	@Override

	protected void map(final ImmutableBytesWritable keyIn, final Result value, final Context context) throws IOException, InterruptedException {

		try {

			final OafRowKeyDecoder rkd = OafRowKeyDecoder.decode(keyIn.get());

			final OafProtos.Oaf oaf = UpdateMerger.mergeBodyUpdates(context, value.getFamilyMap(Bytes.toBytes(rkd.getType().toString())));

			if (oaf == null) {

				return;

			} else {

				emit(context, oaf);

			}

			final Map<byte[], NavigableMap<byte[], byte[]>> row = value.getNoVersionMap();

			for (byte[] cf : row.keySet()) {

				for (Entry<byte[], byte[]> q : value.getFamilyMap(cf).entrySet().stream().filter(e -> {

					final String key = new String(e.getKey());

					boolean skip = key.startsWith("update") || key.equals(DedupUtils.BODY_S);

					if (skip) {

						context.getCounter("export", String.format("skipped %s", StringUtils.substring(key, 0, 6))).increment(1);

					}

					return !skip;

				}).collect(Collectors.toList())) {

					if (new String(q.getValue()).equals("")) {

						context.getCounter("export", "skipped " + new String(cf)).increment(1);

					} else {

						emit(context, OafProtos.Oaf.parseFrom(q.getValue()));

					}

				}

			}

		} catch (final Throwable e) {

			context.getCounter("export", "error: " + e.getClass().getName()).increment(1);

			throw new RuntimeException(e);

		}

	}

	private void emit(Context context, OafProtos.Oaf oaf) throws IOException, InterruptedException {

		Oaf result = null;

		try {

			result = ProtoConverter.convert(oaf);

		} catch (Throwable e) {

			context.getCounter("export", "error:" + e.getClass().getName()).increment(1);

		}

		if (result != null) {

			keyOut.set(result.getClass().getName());

			valueOut.set(objectMapper.writeValueAsString(result));

			final String type = result.getClass().getSimpleName();

			final String namedOutput = type.toLowerCase();

			multipleOutputs.write(namedOutput, keyOut, valueOut, namedOutput + "/" + namedOutput);

			boolean deleted = result.getDataInfo().getDeletedbyinference();

			if (result instanceof Relation) {

				Relation r = (Relation) result;

				String reltype = r.getRelType() + "_" + r.getSubRelType() + "_" + r.getRelClass();

				context.getCounter("export", String.format("%s deleted:%s", reltype, deleted)).increment(1);

			} else if (result instanceof OafEntity) {

				context.getCounter("export", String.format("%s deleted:%s", type, deleted)).increment(1);

			}

		}

	}

}