D-Net

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

import java.io.IOException;

import java.util.List;

import org.apache.commons.lang.StringUtils;

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;

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.hbase.index.config.RelClasses;

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

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

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

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

import eu.dnetlib.data.transform.OafEntityMerger;

import eu.dnetlib.pace.util.DedupConfig;

import eu.dnetlib.pace.util.DedupConfigLoader;

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

	private enum OafPatch {

		rootToEntity, entityToRoot

	}

	private static final boolean WRITE_TO_WAL = false;

	private DedupConfig dedupConf;

	private RelClasses relClasses;

	@Override

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

		super.setup(context);

		dedupConf = DedupConfigLoader.load(context.getConfiguration().get("dedup.wf.conf"));

		System.out.println("dedup buildRoots reducer\n\nwf conf: " + dedupConf.toString());

		final String relClassJson = context.getConfiguration().get("relClasses");

		System.out.println("relClassesJson:\n" + relClassJson);

		relClasses = RelClasses.fromJSon(relClassJson);

		System.out.println("relClasses:\n" + relClasses);

	}

	@Override

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

		// ensures we're dealing with a root, otherwise returns

		if (!DedupUtils.isRoot(key.toString())) {

			System.err.println("aborting DedupBuildRootsReducer, found non-root key: " + key);

			context.getCounter("DedupBuildRootsReducer", "aborted").increment(1);

			return;

		}

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

		final List<Oaf> entities = Lists.newArrayList();

		for (final Oaf oaf : toOaf(values)) {

			switch (oaf.getKind()) {

			case entity:

				entities.add(oaf);

				break;

			case relation:

				handleRels(context, rowkey, oaf, true);

				break;

			default:

				break;

			}

		}

		// build and emit the root body

		final Oaf.Builder builder = OafEntityMerger.merge(dedupConf, key.toString(), entities);

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

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

		} else {

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

		}

		emit(context, rowkey, dedupConf.getEntityType(), DedupUtils.BODY_S, builder.build().toByteArray(), "root");

		// add person rels TODO: remove this hack

		// context.getCounter("hack", "personResult out").increment(personMap.size());

	}

	private Iterable<Oaf> toOaf(final Iterable<ImmutableBytesWritable> values) {

		return Iterables.transform(values, OafHbaseUtils.oafDecoder());

	}

	private void handleRels(final Context context, final byte[] rowkey, final Oaf oaf, final boolean hack) throws IOException, InterruptedException,

	InvalidProtocolBufferException {

		if (hack && checkHack(new String(rowkey), oaf)) {

			context.getCounter("hack", "personResult in").increment(1);

		} else {

			// emit relation from the root to the related entities

			OafDecoder decoder = rootToEntity(rowkey, oaf);

			emit(context, rowkey, decoder.getCFQ(), decoder.relTargetId(), decoder.toByteArray(), "[root -> entity]");

			// emit relation from the related entities to the root

			decoder = entityToRoot(rowkey, oaf);

			final byte[] revKey = Bytes.toBytes(decoder.relSourceId());

			emit(context, revKey, decoder.getCFQ(), new String(rowkey), decoder.toByteArray(), "[entity -> root]");

		}

		// mark relation from the related entities to the duplicate as deleted

		OafDecoder decoder = markDeleted(oaf, true);

		byte[] revKey = Bytes.toBytes(decoder.relSourceId());

		emit(context, revKey, decoder.getCFQ(), decoder.relTargetId(), decoder.toByteArray(), "mark deleted [dup -> entity]");

		// mark relation from the related entities to the duplicate as deleted

		decoder = markDeleted(oaf, false);

		revKey = Bytes.toBytes(decoder.relSourceId());

		emit(context, revKey, decoder.getCFQ(), decoder.relTargetId(), decoder.toByteArray(), "mark deleted [entity -> dup]");

	}

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

			throws IOException, InterruptedException {

		final Put put = new Put(rowkey).add(Bytes.toBytes(family), Bytes.toBytes(qualifier), value);

		put.setWriteToWAL(WRITE_TO_WAL);

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

		context.getCounter(family, label).increment(1);

	}

	// /////////////////

	public boolean checkHack(final String root, final Oaf oaf) {

		boolean res;

		if (dedupConf.getEntityType().equals(Type.result.toString()) && oaf.getRel().getRelType().toString().equals(RelType.personResult.toString())

				&& !md5matches(root, oaf.getRel().getSource())) {

			res = true;

		} else {

			res = false;

		}

		// if (root.equals("50|dedup_wf_001::92f6197ea6f16ae554755aced832fb6f")) {

		// System.out.println("##################");

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

		// System.out.println("source: " + oaf.getRel().getSource());

		// System.out.println("ckeck:  " + res);

		// }

		return res;

	}

	private boolean md5matches(final String id1, final String id2) {

		return id1.replaceAll("^.*\\:\\:", "").equals(id2.replaceAll("^.*\\:\\:", ""));

	}

	private OafDecoder rootToEntity(final byte[] rootRowkey, final Oaf rel) {

		return patchRelations(rootRowkey, rel, OafPatch.rootToEntity);

	}

	private OafDecoder entityToRoot(final byte[] rootRowkey, final Oaf rel) {

		return patchRelations(rootRowkey, rel, OafPatch.entityToRoot);

	}

	private OafDecoder markDeleted(final Oaf rel, final boolean reverse) {

		return deleteRelations(rel, reverse);

	}

	// patches relation objects setting the source field with the root id

	private OafDecoder patchRelations(final byte[] rootRowkey, final Oaf rel, final OafPatch patchKind) {

		final OafRelDecoder decoder = OafRelDecoder.decode(rel.getRel());

		final Oaf.Builder builder = Oaf.newBuilder(rel);

		builder.getDataInfoBuilder().setInferred(true).setDeletedbyinference(false);

		switch (patchKind) {

		case rootToEntity:

			// builder.getDataInfoBuilder().setInferenceprovenance("dedup (BuildRoots p:rootToEntity)");

			builder.getRelBuilder().setSource(new String(rootRowkey));

			break;

		case entityToRoot:

			builder.setRel(decoder.setClassId(relClasses.getInverse(rel.getRel().getRelClass())));

			// builder.getDataInfoBuilder().setInferenceprovenance("dedup (BuildRoots p:entityToRoot)");

			builder.getRelBuilder().setSource(builder.getRel().getTarget());

			builder.getRelBuilder().setTarget(new String(rootRowkey));

			break;

		default:

			break;

		}

		return OafDecoder.decode(builder.build());

	}

	private OafDecoder deleteRelations(final Oaf rel, final boolean reverse) {

		final Oaf.Builder builder = Oaf.newBuilder(rel);

		// builder.getDataInfoBuilder().setInferenceprovenance("dedup (BuildRoots d: " + reverse + ")");

		builder.getDataInfoBuilder().setDeletedbyinference(true);

		if (reverse) {

			final OafRelDecoder decoder = OafRelDecoder.decode(rel.getRel());

			builder.setRel(decoder.setClassId(relClasses.getInverse(rel.getRel().getRelClass())));

			// swap source and target

			final String tmp = builder.getRel().getSource();

			builder.getRelBuilder().setSource(builder.getRel().getTarget());

			builder.getRelBuilder().setTarget(tmp);

		}

		return OafDecoder.decode(builder.build());

	}

	private String lpad(final int s) {

		return StringUtils.leftPad(String.valueOf(s), 5);

	}

}