D-Net

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

import java.io.IOException;

import java.util.Map;

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;

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;

public class DedupDeleteRelMapper 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());

		deleteRels(rowkey, context, value, DedupUtils.getSimilarityCFBytes(type));

		deleteRels(rowkey, context, value, DedupUtils.getDedupCF_mergedInBytes(type));

		deleteRels(rowkey, context, value, DedupUtils.getDedupCF_mergesBytes(type));

	}

	private void deleteRels(final ImmutableBytesWritable rowkey, final Context context, final Result value, final byte[] cf)

			throws IOException, InterruptedException {

		final Map<byte[], byte[]> rels = value.getFamilyMap(cf);

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

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

			final Delete delete = new Delete(row);

			delete.setWriteToWAL(JobParams.WRITE_TO_WAL);

			delete.deleteFamily(cf);

			outKey.set(row);

			context.write(outKey, delete);

			context.getCounter(dedupConf.getWf().getEntityType(), new String(cf) + " deleted").increment(rels.size());

		}

	}

}