D-Net

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

import java.io.IOException;

import java.nio.ByteBuffer;

import java.util.Set;

import com.google.common.collect.Sets;

import eu.dnetlib.data.mapreduce.JobParams;

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

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

import eu.dnetlib.data.proto.KindProtos.Kind;

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

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

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

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

import eu.dnetlib.data.transform.xml.AbstractDNetXsltFunctions;

import eu.dnetlib.pace.config.DedupConfig;

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;

/**

 * 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) {

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

		log.info("dedup findRoots mapper\nwf conf: " + 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) throws IOException, InterruptedException {

		final Set<String> set = Sets.newHashSet();

		for(VertexWritable v : values) {

			for(Text t : v.getEdges()) {

				set.add(t.toString());

			}

		}

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

		for(String q : set) {

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

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

			InterruptedException {

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

		put.setWriteToWAL(JobParams.WRITE_TO_WAL);

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

	}

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

	}

}