D-Net

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

import java.io.IOException;

import java.util.Arrays;

import java.util.List;

import org.apache.hadoop.hbase.KeyValue;

import org.apache.hadoop.hbase.client.Put;

import org.apache.hadoop.hbase.client.Result;

import org.apache.hadoop.hbase.io.ImmutableBytesWritable;

import org.apache.hadoop.hbase.mapreduce.TableMapper;

import org.apache.hadoop.hbase.util.Bytes;

import org.apache.hadoop.io.Text;

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

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

import eu.dnetlib.pace.util.DedupConfig;

import eu.dnetlib.pace.util.DedupConfigLoader;

public class DedupGrouperMapper extends TableMapper<Text, Put> {

	private static final boolean WRITE_TO_WAL = false;

	public static final String COUNTER_GROUP = "dedup.grouper";

	public static final String COUNTER_NAME = "written.rels";

	private Text rowKey;

	private DedupConfig dedupConf;

	@Override

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

		rowKey = new Text();

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

	}

	@Override

	protected void map(final ImmutableBytesWritable keyIn, final Result value, final Context context) throws IOException, InterruptedException {

		final List<KeyValue> kvList = value.list();

		//System.out.println("Grouper mapping " + kvList.size() + " rels for key: " + new String(keyIn.copyBytes()));

		for (KeyValue n : kvList) {

			for (KeyValue j : kvList) {

				byte[] nq = n.getQualifier();

				byte[] jq = j.getQualifier();

				if (!Arrays.equals(nq, jq)) {

					Put put = new Put(nq).add(DedupUtils.getSimilarityCFBytes(Type.valueOf(dedupConf.getEntityType())), jq, Bytes.toBytes(""));

					put.setWriteToWAL(WRITE_TO_WAL);

					rowKey.set(nq);

					context.write(rowKey, put);

					context.getCounter(COUNTER_GROUP, COUNTER_NAME).increment(1);

				}

			}

		}

	}

}