D-Net

package eu.dnetlib.data.mapreduce.hbase.statsExport.drivers;

import java.io.InputStream;

import java.util.Properties;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.mapreduce.Job;

import org.apache.log4j.Logger;

import eu.dnetlib.data.mapreduce.hbase.statsExport.StatsJob;

/**

 * @author eri

 * 

 */

public class StatsJobDriver {

	private Logger log = Logger.getLogger(this.getClass());

	private Configuration configuration = new Configuration();

	private StatsJob statsJob;

	private Job job;

	private Properties props;

	public void init() throws Exception {

		InputStream file = ClassLoader.getSystemResourceAsStream("eu/dnetlib/data/mapreduce/hbase/statsExport/StatsProperties");

		props = new Properties();

		props.load(file);

		file.close();

		configuration.set("stats.delim", props.getProperty("Stats.delimCharacter"));

		configuration.set("hbase.mapreduce.inputtable", props.getProperty("Stats.HbaseSourceTable"));

		if (!props.getProperty("Stats.outputPath").endsWith("/")) {

			props.setProperty("Stats.outputPath", props.getProperty("Stats.outputPath") + "/");

		}

		configuration.set("mapred.output.dir", props.getProperty("Stats.outputPath"));

		configuration.set("stats.nullNum", props.getProperty("Stats.nullNumericField"));

		configuration.set("stats.nullString", props.getProperty("Stats.nullStringField"));

		job = new Job(configuration);

		statsJob = new StatsJob();

		statsJob.setJobDetails(job, props);

	}

	public void run() throws Exception {

		long startTime = System.currentTimeMillis();

		long endtime;

		double totalTime1;

		// TODO here, start the Stats Export Job: Read from HBASE, and dump to

		// HDFS

		boolean completionStatus = job.waitForCompletion(true);

		if (completionStatus) {

			endtime = System.currentTimeMillis();

			totalTime1 = (endtime - startTime) / 60000;

			log.info("Time taken for MR Export : " + totalTime1 + " minutes");

		} else {

			log.error("JOB FAILED :");

		}

	}

}