D-Net

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

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

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;

/**

 * 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 Gson g;

	@Override

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

		super.setup(context);

		keyOut = new Text();

		valueOut = new Text();

		multipleOutputs = new MultipleOutputs(context);

		g = new Gson();

	}

	@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 : row.get(cf).entrySet()) {

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

					if(key.startsWith("update") || DedupUtils.BODY_S.equals(key)) {

						continue;

					}

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

						context.getCounter("export", "skipped relation " + 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(g.toJson(result));

			final String namedOutput = result.getClass().getSimpleName().toLowerCase();

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

			context.getCounter("export", result.getClass().getName()).increment(1);

		}

	}

}