D-Net

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

import java.io.IOException;

import java.util.Properties;

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

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

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

import org.apache.hadoop.mapreduce.Job;

import eu.dnetlib.data.mapreduce.JobParams;

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

public class HBaseResetJob extends AbstractHBaseMapReduceJob {

	@Override

	protected Job setJobDetails(Job job, Properties p) throws Exception {

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

		job.setOutputFormatClass(TableOutputFormat.class);

		job.setMapOutputKeyClass(Text.class);

		job.setMapOutputValueClass(Mutation.class);

		job.setNumReduceTasks(0);

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

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

		return job;

	}

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

		try {

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

		} catch (IOException e) {

			throw new RuntimeException(e);

		}

	}

	private Scan getScan(Properties p) {

		Scan scan = new Scan();

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

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

		return scan;

	}

}