D-Net

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

import java.io.IOException;

import java.util.Properties;

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

import org.apache.hadoop.hbase.filter.PrefixFilter;

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

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

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

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

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Job;

import eu.dnetlib.data.mapreduce.JobParams;

import eu.dnetlib.data.mapreduce.hbase.AbstractHBaseMapReduceJob;

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

import eu.dnetlib.pace.util.DedupConfig;

import eu.dnetlib.pace.util.DedupConfigLoader;

public class DedupPersonJob extends AbstractHBaseMapReduceJob {

	@Override

	public Job setJobDetails(final Job job, final Properties p) {

		initJob(job, getScan(p), p.getProperty(JobParams.HBASE_SOURCE_TABLE));

		job.setOutputFormatClass(TableOutputFormat.class);

		job.setNumReduceTasks(1000);

		job.getConfiguration().setBoolean("mapred.compress.map.output", true);

		job.getConfiguration().setBoolean("mapred.reduce.tasks.speculative.execution", false);

		job.getConfiguration().setBoolean("mapred.map.tasks.speculative.execution", false);

		job.getConfiguration().setBoolean("mapreduce.map.speculative", false);

		job.getConfiguration().setBoolean("mapreduce.reduce.speculative", false);

		// TODO under test

		// job.setCombinerClass(DedupPersonReducer.class);

		job.getConfiguration().set(TableOutputFormat.OUTPUT_TABLE, p.getProperty(JobParams.HBASE_TARGET_TABLE));

		return job;

	}

	private Scan getScan(final Properties p) {

		Scan scan = new Scan();

		scan.setCaching(500); // 1 is the default in Scan, which will be bad for MapReduce jobs

		scan.setCacheBlocks(false); // don't set to true for MR jobs

		DedupConfig dedupConfig = DedupConfigLoader.load(p.getProperty("dedup.wf.conf"));

		Type type = Type.valueOf(dedupConfig.getEntityType());

		scan.setFilter(new PrefixFilter(Bytes.toBytes(type.getNumber())));

		// scan.addFamily(dedupConfig.getEntityNameBytes());

		return scan;

	}

	private void initJob(final Job job, final Scan scan, final String sourceTable) {

		try {

			TableMapReduceUtil.initTableMapperJob(sourceTable, scan, DedupPersonMapper.class, Text.class, ImmutableBytesWritable.class, job);

			TableMapReduceUtil.initTableReducerJob(sourceTable, DedupPersonReducer.class, job);

		} catch (IOException e) {

			throw new RuntimeException(e);

		}

	}

}