D-Net

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

import java.io.IOException;

import java.util.Queue;

import com.google.common.collect.Iterables;

import com.google.common.collect.Lists;

import com.google.protobuf.InvalidProtocolBufferException;

import eu.dnetlib.data.mapreduce.JobParams;

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

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

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

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

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;

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

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

import eu.dnetlib.pace.config.DedupConfig;

import org.apache.commons.lang.StringUtils;

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

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

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

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

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

import org.apache.hadoop.io.Text;

public class SimpleDedupPersonReducer extends TableReducer<Text, ImmutableBytesWritable, ImmutableBytesWritable> {

	private static final int MAX_Q_SIZE = 3000;

	private DedupConfig dedupConf;

	public static String findMin(final Iterable<String> keys) {

		String min = Iterables.getFirst(keys, null);

		for (final String iq : keys) {

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

				min = iq;

			}

		}

		return min;

	}

	@Override

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

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

	}

	@Override

	protected void reduce(final Text key, final Iterable<ImmutableBytesWritable> values, final Context context) throws IOException, InterruptedException {

		try {

			final Queue<OafDecoder> q = prepare(key, values, context);

			if (q.size() > 1) {

				if (q.size() < JobParams.MAX_COUNTERS) {

					context.getCounter(dedupConf.getWf().getEntityType() + " root group size", lpad(q.size())).increment(1);

				} else {

					context.getCounter(dedupConf.getWf().getEntityType() + " root group size", "> " + JobParams.MAX_COUNTERS).increment(1);

				}

				final String min = findMin(Iterables.transform(q, eu.dnetlib.data.transform.OafUtils.idDecoder()));

				if (min == null) {

					context.getCounter(dedupConf.getWf().getEntityType(), "unable to find min").increment(1);

					return;

				}

				final String rootId = DedupUtils.newId(min, dedupConf.getWf().getDedupRun());

				while (!q.isEmpty()) {

					markDuplicate(context, rootId, q.remove());

				}

			} else {

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

			}

		} catch (final Throwable e) {

			System.out.println("GOT EX " + e);

			e.printStackTrace(System.err);

			context.getCounter(dedupConf.getWf().getEntityType(), e.getClass().toString()).increment(1);

		}

	}

	private Queue<OafDecoder> prepare(final Text key, final Iterable<ImmutableBytesWritable> values, final Context context) {

		final Queue<OafDecoder> q = Lists.newLinkedList();

		for (final OafDecoder decoder : Iterables.transform(values, OafHbaseUtils.decoder())) {

			q.add(decoder);

			if (q.size() >= MAX_Q_SIZE) {

				//COMMENTED OUT TO AVOID THE JOB FAILS BECAUSE OF MAX NUMBER OF COUNTERS REACHED

				//context.getCounter("[" + key.toString() + "]", "size > " + MAX_Q_SIZE).increment(1);

				break;

			}

		}

		return q;

	}

	private void markDuplicate(final Context context, final String rootId, final OafDecoder decoder) throws InvalidProtocolBufferException, IOException,

			InterruptedException {

		final Oaf.Builder builder = Oaf.newBuilder(decoder.getOaf());

		builder.getDataInfoBuilder().setDeletedbyinference(true).setInferenceprovenance(dedupConf.getWf().getConfigurationId());

		final Oaf oaf = builder.build();

		final byte[] oafId = Bytes.toBytes(oaf.getEntity().getId());

		// writes the body, marked as deleted

		final String entityName = dedupConf.getWf().getEntityType();

		emit(context, oafId, entityName, DedupUtils.BODY_B, oaf.toByteArray());

		context.getCounter(entityName, "marked as deleted").increment(1);

		// writes the dedupRels in both directions

		final Type entityType = Type.valueOf(entityName);

		final byte[] rowkey = Bytes.toBytes(rootId);

		final String merges = DedupUtils.getDedupCF_merges(entityType);

		emit(context, rowkey, merges, oafId, buildRel(rowkey, oafId, Dedup.RelName.merges));

		final String mergedIn = DedupUtils.getDedupCF_mergedIn(entityType);

		emit(context, oafId, mergedIn, rowkey, buildRel(oafId, rowkey, Dedup.RelName.isMergedIn));

		context.getCounter(entityName, merges).increment(1);

		context.getCounter(entityName, mergedIn).increment(1);

	}

	private void emit(final Context context, final byte[] rowkey, final String family, final byte[] qualifier, final byte[] value) throws IOException,

			InterruptedException {

		final Put put = new Put(OafRowKeyDecoder.decode(rowkey).getKey().getBytes());

		put.setDurability(Durability.USE_DEFAULT);

		put.add(Bytes.toBytes(family), qualifier, value);

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

	}

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

		final OafRel.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 String lpad(final int s) {

		return StringUtils.leftPad(String.valueOf(s), String.valueOf(MAX_Q_SIZE).length());

	}

}