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.lang3.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 person:

//				process(q, context);

//				break;

//			case publication:

//			case dataset:

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

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

//		emitRel(context, target, rowKey);

//	}

//

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

//		final Put put = new Put(from).add(DedupUtils.getSimilarityCFBytes(Type.valueOf(dedupConf.getWf().getEntityType())), to, Bytes.toBytes(""));

//		put.setWriteToWAL(JobParams.WRITE_TO_WAL);

//		ibw.set(from);

//		context.write(ibw, put);

//	}

//}