D-Net

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

import java.io.IOException;

import java.util.*;

import com.google.common.collect.Lists;

import eu.dnetlib.data.mapreduce.JobParams;

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

import eu.dnetlib.data.proto.RelTypeProtos.SubRelType;

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

import eu.dnetlib.pace.clustering.NGramUtils;

import eu.dnetlib.pace.config.DedupConfig;

import eu.dnetlib.pace.distance.PaceDocumentDistance;

import eu.dnetlib.pace.distance.eval.ScoreResult;

import eu.dnetlib.pace.model.Field;

import eu.dnetlib.pace.model.MapDocument;

import eu.dnetlib.pace.model.MapDocumentComparator;

import eu.dnetlib.pace.model.MapDocumentSerializer;

import org.apache.commons.lang.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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 DedupReducer extends TableReducer<Text, ImmutableBytesWritable, ImmutableBytesWritable> {

	private static final Log log = LogFactory.getLog(DedupReducer.class);

	private DedupConfig dedupConf;

	private ImmutableBytesWritable ibw;

	@Override

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

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

		ibw = new ImmutableBytesWritable();

		log.info("dedup reduce phase \npace conf: " + dedupConf.toString());

	}

	@Override

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

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

		if (q.size() > 1) {

			log.info("reducing key: '" + key + "' records: " + q.size());

			switch (Type.valueOf(dedupConf.getWf().getEntityType())) {

			case result:

				process(simplifyQueue(q, key.toString(), context), context);

				break;

			case organization:

				process(q, context);

				break;

			default:

				throw new IllegalArgumentException("process not implemented for type: " + dedupConf.getWf().getEntityType());

			}

		} else {

			context.getCounter(dedupConf.getWf().getEntityType(), "records per hash key = 1").increment(1);

		}

	}

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

		final Queue<MapDocument> queue = new PriorityQueue<MapDocument>(100, new MapDocumentComparator(dedupConf.getWf().getOrderField()));

		final Set<String> seen = new HashSet<String>();

		final int queueMaxSize = dedupConf.getWf().getQueueMaxSize();

		int count = 0;

		boolean logged = false;

		for (final ImmutableBytesWritable i : values) {

			count++;

			if (queue.size() <= queueMaxSize) {

				final MapDocument doc = MapDocumentSerializer.decode(i.copyBytes());

				final String id = doc.getIdentifier();

				if (!seen.contains(id)) {

					seen.add(id);

					queue.add(doc);

				}

			} else {

				if (!logged) {

					// context.getCounter("ngram size > " + LIMIT, "'" + key.toString() + "', --> " + context.getTaskAttemptID()).increment(1);

					context.getCounter("ngram size > " + queueMaxSize, "N").increment(1);

					log.info("breaking out after limit (" + queueMaxSize + ") for ngram '" + key + "'");

					logged = true;

				}

			}

		}

		log.info(String.format("cluster key '%s' size '%s'", key, count));

		return queue;

	}

	private Queue<MapDocument> simplifyQueue(final Queue<MapDocument> queue, final String ngram, final Context context) {

		final Queue<MapDocument> q = new LinkedList<MapDocument>();

		String fieldRef = "";

		final List<MapDocument> tempResults = Lists.newArrayList();

		while (!queue.isEmpty()) {

			final MapDocument result = queue.remove();

			final String orderFieldName = dedupConf.getWf().getOrderField();

			final Field orderFieldValue = result.values(orderFieldName);

			if (!orderFieldValue.isEmpty()) {

				final String field = NGramUtils.cleanupForOrdering(orderFieldValue.stringValue());

				if (field.equals(fieldRef)) {

					tempResults.add(result);

				} else {

					populateSimplifiedQueue(q, tempResults, context, fieldRef, ngram);

					tempResults.clear();

					tempResults.add(result);

					fieldRef = field;

				}

			} else {

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

			}

		}

		populateSimplifiedQueue(q, tempResults, context, fieldRef, ngram);

		return q;

	}

	private void populateSimplifiedQueue(final Queue<MapDocument> q,

			final List<MapDocument> tempResults,

			final Context context,

			final String fieldRef,

			final String ngram) {

		if (tempResults.size() < dedupConf.getWf().getGroupMaxSize()) {

			q.addAll(tempResults);

		} else {

			context.getCounter(dedupConf.getWf().getEntityType(),

					"Skipped records for count(" + dedupConf.getWf().getOrderField() + ") >= " + dedupConf.getWf().getGroupMaxSize())

					.increment(tempResults.size());

			log.info("Skipped field: " + fieldRef + " - size: " + tempResults.size() + " - ngram: " + ngram);

		}

	}

	private void process(final Queue<MapDocument> queue, final Context context) throws IOException, InterruptedException {

		final PaceDocumentDistance algo = new PaceDocumentDistance();

		while (!queue.isEmpty()) {

			final MapDocument pivot = queue.remove();

			final String idPivot = pivot.getIdentifier();

			final Field fieldsPivot = pivot.values(dedupConf.getWf().getOrderField());

			final String fieldPivot = (fieldsPivot == null) || fieldsPivot.isEmpty() ? null : fieldsPivot.stringValue();

			if (fieldPivot != null) {

				// System.out.println(idPivot + " --> " + fieldPivot);

				int i = 0;

				for (final MapDocument curr : queue) {

					final String idCurr = curr.getIdentifier();

					if (mustSkip(idCurr)) {

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

						break;

					}

					if (i > dedupConf.getWf().getSlidingWindowSize()) {

						break;

					}

					final Field fieldsCurr = curr.values(dedupConf.getWf().getOrderField());

					final String fieldCurr = (fieldsCurr == null) || fieldsCurr.isEmpty() ? null : fieldsCurr.stringValue();

					if (!idCurr.equals(idPivot) && (fieldCurr != null)) {

						final ScoreResult sr = similarity(algo, pivot, curr);

						emitOutput(sr, idPivot, idCurr, context);

						i++;

					}

				}

			}

		}

	}

	private void emitOutput(final ScoreResult sr, final String idPivot, final String idCurr, final Context context) throws IOException, InterruptedException {

		final double d = sr.getScore();

		if (d >= dedupConf.getWf().getThreshold()) {

			writeSimilarity(context, idPivot, idCurr, d);

			context.getCounter(dedupConf.getWf().getEntityType(), SubRelType.dedupSimilarity.toString() + " (x2)").increment(1);

		} else {

			context.getCounter(dedupConf.getWf().getEntityType(), "d < " + dedupConf.getWf().getThreshold()).increment(1);

		}

	}

	private ScoreResult similarity(final PaceDocumentDistance algo, final MapDocument a, final MapDocument b) {

		try {

			return algo.between(a, b, dedupConf);

		} catch(Throwable e) {

			log.error(String.format("\nA: %s\n----------------------\nB: %s", a, b), e);

			throw new IllegalArgumentException(e);

		}

	}

	private boolean mustSkip(final String idPivot) {

		return dedupConf.getWf().getSkipList().contains(getNsPrefix(idPivot));

	}

	private String getNsPrefix(final String id) {

		return StringUtils.substringBetween(id, "|", "::");

	}

	private void writeSimilarity(final Context context, final String idPivot, final String id, final double d) throws IOException, InterruptedException {

		final byte[] rowKey = Bytes.toBytes(idPivot);

		final byte[] target = Bytes.toBytes(id);

		//log.info("writing similarity: " + idPivot + " <-> " + id);

		emitRel(context, rowKey, target, d);

		emitRel(context, target, rowKey, d);

	}

	private void emitRel(final Context context, final byte[] from, final byte[] to, final double d) throws IOException, InterruptedException {

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

		//final OafRel.Builder rel = DedupUtils.getDedupSimilarity(dedupConf, new String(from), new String(to));

		//final Oaf.Builder oaf = DedupUtils.buildRel(dedupConf, rel, d);

		//final Put put = new Put(from).add(DedupUtils.getSimilarityCFBytes(type), to, oaf.build().toByteArray());

		final Put put = new Put(from).add(DedupUtils.getSimilarityCFBytes(type), to, Bytes.toBytes(""));

		put.setWriteToWAL(JobParams.WRITE_TO_WAL);

		ibw.set(from);

		context.write(ibw, put);

	}

}