D-Net

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

import java.io.IOException;

import java.nio.ByteBuffer;

import java.util.Map;

import com.google.protobuf.InvalidProtocolBufferException;

import eu.dnetlib.data.mapreduce.JobParams;

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

import eu.dnetlib.data.proto.DedupProtos.Dedup;

import eu.dnetlib.data.proto.KindProtos.Kind;

import eu.dnetlib.data.proto.OafProtos.Oaf;

import eu.dnetlib.data.proto.OafProtos.OafRel.Builder;

import eu.dnetlib.data.proto.TypeProtos.Type;

import eu.dnetlib.data.transform.xml.AbstractDNetXsltFunctions;

import eu.dnetlib.pace.config.DedupConfig;

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

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

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;

public class DedupFindRootsMapper extends TableMapper<ImmutableBytesWritable, Put> {

	private DedupConfig dedupConf;

	@Override

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

		dedupConf = DedupConfig.load(context.getConfiguration().get(JobParams.DEDUP_CONF));

		System.out.println("dedup findRoots mapper\nwf conf: " + dedupConf.toString());

	}

	@Override

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

		// System.out.println("Find root mapping: " + new String(rowkey.copyBytes()));

		final Type type = Type.valueOf(dedupConf.getWf().getEntityType());

		final Map<byte[], byte[]> similarRels = value.getFamilyMap(DedupUtils.getSimilarityCFBytes(type));

		if ((similarRels != null) && !similarRels.isEmpty()) {

			final ByteBuffer min = findMin(ByteBuffer.wrap(rowkey.get()), similarRels.keySet());

			final byte[] row = rowkey.copyBytes();

			final byte[] root = DedupUtils.newIdBytes(min, dedupConf.getWf().getDedupRun());

			// System.out.println("Found root: " + new String(root));

			emitDedupRel(context, DedupUtils.getDedupCF_mergedInBytes(type), row, root, buildRel(row, root, Dedup.RelName.isMergedIn));

			emitDedupRel(context, DedupUtils.getDedupCF_mergesBytes(type), root, row, buildRel(root, row, Dedup.RelName.merges));

			context.getCounter(dedupConf.getWf().getEntityType(), "dedupRel (x2)").increment(1);

			// marks the original body deleted

			emitBody(context, row, value.getValue(Bytes.toBytes(dedupConf.getWf().getEntityType()), DedupUtils.BODY_B));

		} else {

			context.getCounter(dedupConf.getWf().getEntityType(), "row not in similarity mesh").increment(1);

		}

	}

	private ByteBuffer findMin(ByteBuffer min, final Iterable<byte[]> keys) {

		for (final byte[] q : keys) {

			final ByteBuffer iq = ByteBuffer.wrap(q);

			if (min.compareTo(iq) > 0) {

				min = iq;

			}

		}

		return min;

	}

	private void emitBody(final Context context, final byte[] row, final byte[] body) throws InvalidProtocolBufferException, IOException, InterruptedException {

		if (body == null) {

			context.getCounter(dedupConf.getWf().getEntityType(), "missing body").increment(1);

			System.err.println("missing body: " + new String(row));

			return;

		}

		final Oaf prototype = Oaf.parseFrom(body);

		if (prototype.getDataInfo().getDeletedbyinference()) {

			context.getCounter(dedupConf.getWf().getEntityType(), "bodies already deleted").increment(1);

		} else {

			final Oaf.Builder oafRoot = Oaf.newBuilder(prototype);

			oafRoot.getDataInfoBuilder().setDeletedbyinference(true).setInferred(true).setInferenceprovenance(dedupConf.getWf().getConfigurationId());

			final byte[] family = Bytes.toBytes(dedupConf.getWf().getEntityType());

			final Put put = new Put(row).add(family, DedupUtils.BODY_B, oafRoot.build().toByteArray());

			put.setDurability(Durability.USE_DEFAULT);

			context.write(new ImmutableBytesWritable(row), put);

			context.getCounter(dedupConf.getWf().getEntityType(), "bodies marked deleted").increment(1);

		}

	}

	private byte[] buildRel(final byte[] from, final byte[] to, final Dedup.RelName relClass) {

		final Builder oafRel = DedupUtils.getDedup(dedupConf, new String(from), new String(to), relClass);

		final Oaf oaf =

				Oaf.newBuilder()

						.setKind(Kind.relation)

						.setLastupdatetimestamp(System.currentTimeMillis())

						.setDataInfo(

								AbstractDNetXsltFunctions.getDataInfo(null, "", "0.8", false, true).setInferenceprovenance(

										dedupConf.getWf().getConfigurationId())).setRel(oafRel)

				.build();

		return oaf.toByteArray();

	}

	private void emitDedupRel(final Context context, final byte[] cf, final byte[] from, final byte[] to, final byte[] value) throws IOException,

			InterruptedException {

		final Put put = new Put(from).add(cf, to, value);

		put.setWriteToWAL(JobParams.WRITE_TO_WAL);

		context.write(new ImmutableBytesWritable(from), put);

	}

}