D-Net

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

import java.io.IOException;

import java.util.Map;

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

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.io.Writable;

import eu.dnetlib.data.mapreduce.JobParams;

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

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

import eu.dnetlib.pace.config.DedupConfig;

public class DedupDeleteSimRelMapper extends TableMapper<ImmutableBytesWritable, Writable> {

	private DedupConfig dedupConf;

	private ImmutableBytesWritable outKey;

	@Override

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

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

		System.out.println("dedup findRoots mapper\nwf conf: " + dedupConf.toString());

		outKey = new ImmutableBytesWritable();

	}

	@Override

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

		// System.out.println("Find root mapping: " + new String(rowkey.copyBytes()));

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

		final Map<byte[], byte[]> similarRels = value.getFamilyMap(DedupUtils.getSimilarityCFBytes(type));

		if ((similarRels != null) && !similarRels.isEmpty()) {

			final byte[] row = rowkey.copyBytes();

			final Delete delete = new Delete(row);

			for (final byte[] q : similarRels.keySet()) {

				delete.deleteColumns(DedupUtils.getSimilarityCFBytes(type), q);

			}

			outKey.set(row);

			context.write(outKey, delete);

			context.getCounter(dedupConf.getWf().getEntityType(), "similarity deleted").increment(similarRels.size());

		} else {

			context.getCounter(dedupConf.getWf().getEntityType(), "row not in similarity mesh").increment(1);

		}

	}

}