D-Net

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

import java.io.IOException;

import java.nio.ByteBuffer;

import eu.dnetlib.data.mapreduce.JobParams;

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

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

import eu.dnetlib.data.proto.DedupProtos.Dedup;

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

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

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

import eu.dnetlib.pace.config.DedupConfig;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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;

/**

 * Created by claudio on 15/10/15.

 */

public class ConnectedComponentsReducer extends TableReducer<Text, VertexWritable, ImmutableBytesWritable> {

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

	private DedupConfig dedupConf;

	private byte[] cfMergedIn;

	private byte[] cfMerges;

	@Override

	protected void setup(final Context context) {

		final String conf = context.getConfiguration().get(JobParams.DEDUP_CONF);

		log.info("dedup configuration string\n" +conf);

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

		log.info("dedup configuration parsed\n" + dedupConf.toString());

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

		cfMergedIn = DedupUtils.getDedupCF_mergedInBytes(type);

		cfMerges = DedupUtils.getDedupCF_mergesBytes(type);

	}

	@Override

	protected void reduce(Text key, Iterable<VertexWritable> values, Context context) {

		final byte[] root = DedupUtils.newIdBytes(ByteBuffer.wrap(Bytes.toBytes(key.toString())), dedupConf.getWf().getDedupRun());

		StreamUtils.toStream(values.iterator())

				.limit(dedupConf.getWf().getGroupMaxSize())  // might cause timeouts in case of large number of items

				.flatMap(v -> v.getEdges().stream())

				.forEach(q -> {

					final byte[] qb = Bytes.toBytes(q.toString());

					emitDedupRel(context, cfMergedIn, qb, root, buildRel(qb, root, Dedup.RelName.isMergedIn));

					emitDedupRel(context, cfMerges, root, qb, buildRel(root, qb, Dedup.RelName.merges));

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

				});

	}

	private void emitDedupRel(final Context context, final byte[] cf, final byte[] from, final byte[] to, final byte[] value) {

		final Put put = new Put(from).add(cf, to, value);

		put.setDurability(Durability.USE_DEFAULT);

		try {

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

		} catch (IOException | InterruptedException e) {

			throw new RuntimeException(e);

		}

	}

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

		final OafRel.Builder oafRef = DedupUtils.getDedup(dedupConf, new String(from), new String(to), relClass);

		final Oaf oaf = DedupUtils.buildRel(dedupConf, oafRef, 0.8).build();

		return oaf.toByteArray();

	}

}