D-Net

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

import java.io.IOException;

import java.util.List;

import com.google.common.collect.Iterables;

import com.google.common.collect.Lists;

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.transform.OafEntityMerger;

import eu.dnetlib.pace.config.DedupConfig;

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;

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

	private enum OafPatch {

		rootToEntity, entityToRoot

	}

	private DedupConfig dedupConf;

	private RelClasses relClasses;

	@Override

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

		super.setup(context);

		dedupConf = DedupConfig.load(context.getConfiguration().get(JobParams.DEDUP_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);

				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.getWf().getEntityType() + " root group size", lpad(entities.size())).increment(1);

		} else {

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

		}

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

	}

	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) throws IOException, InterruptedException {

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

		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

		decoder = markDeleted(oaf, true);

		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(JobParams.WRITE_TO_WAL);

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

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

	}

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

	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 String id = new String(rootRowkey);

		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(id);

			break;

		case entityToRoot:

			final String relClass = rel.getRel().getRelClass();

			/*

			if(StringUtils.isBlank(relClass)) {

				throw new IllegalStateException(String.format("missing relation term for %s in row %s", rel.getRel().getRelType().name(), id));

			}

			*/

			final String inverse = relClasses.getInverse(relClass);

			if(StringUtils.isBlank(inverse)) {

				throw new IllegalStateException(String.format("missing inverse relation for %s in row %s", relClass, id));

			}

			builder.setRel(decoder.setClassId(inverse));

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

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

			builder.getRelBuilder().setTarget(id);

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

	}

}