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.NullWritable;

import org.apache.hadoop.io.Text;

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

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;

import java.util.stream.Stream;

/**

 * 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);

			}

			Stream.of(value.raw())

					.filter(kv -> {

						final String q = Bytes.toString(kv.getQualifier());

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

						if (skip) {

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

						}

						return !skip;

					})

					.filter(kv -> !"".equals(Bytes.toString(kv.getValue())))

					.map(kv -> kv.getValue())

					.forEach(v -> {

						try {

							emit(context, OafProtos.Oaf.parseFrom(v));

						} catch (IOException | InterruptedException e) {

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

							throw new RuntimeException(e);

						}

					});

		} 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) {

			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);

			}

		}

	}

}