D-Net

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

import java.io.InputStream;

import java.util.Map.Entry;

import java.util.Properties;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.FileSystem;

import org.apache.hadoop.fs.Path;

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

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

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

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

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

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Job;

import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;

import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;

import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

import org.apache.log4j.Logger;

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

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

public class StatsJob extends AbstractHBaseMapReduceJob {

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

	private FileSystem hdfs;

	private Properties tableMappings = new Properties();

	private String TABLE_MAP_PATH = "eu/dnetlib/data/mapreduce/hbase/statsExport/exportTables";

	@Override

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

		InputStream file = ClassLoader.getSystemResourceAsStream(TABLE_MAP_PATH);

		tableMappings.load(file);

		file.close();

		// TODO dnet config - only if working directly on cluster

		dnetConfig(job);

		if (p != null) {

			for (Entry<Object, Object> prop : p.entrySet()) {

				job.getConfiguration().set(prop.getKey().toString(), prop.getValue().toString());

			}

		}

		job.setJarByClass(StatsJob.class);

		job.setMapperClass(StatsMapper.class);

		job.setReducerClass(StatsReducer.class);

		job.setInputFormatClass(TableInputFormat.class);

		job.setMapOutputKeyClass(Text.class);

		job.setMapOutputValueClass(ImmutableBytesWritable.class);

		//

		// TODO here how many m/r tasks????

		job.setNumReduceTasks(10);

		this.hdfs = FileSystem.get(new Configuration());

		job.getConfiguration().set(TableInputFormat.INPUT_TABLE, job.getConfiguration().get("hbase.mapreduce.inputtable"));

		job.setOutputValueClass(SequenceFileOutputFormat.class);

		// TODO

		hdfs.delete(new Path(job.getConfiguration().get("mapred.output.dir")), 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);

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

		Scan scan = new Scan();

		// TODO add all column families here to get correct results?

		// scan.addFamily(Bytes.toBytes("result"));

		// scan.addFamily(Bytes.toBytes("personResult_authorship_hasAuthor"));

		// scan.addFamily(Bytes.toBytes("resultResult_publicationDataset_isRelatedTo"));

		scan.addFamily(Bytes.toBytes("datasource"));

		// scan.addFamily(Bytes.toBytes("datasourceOrganization_provision_provides"));

		scan.setCaching(200);

		for (Entry<Object, Object> e : tableMappings.entrySet()) {

			String type = (String) e.getKey();

			log.info("Creating MultipleOutput : " + type.toString());

			MultipleOutputs.addNamedOutput(job, type.toString(), TextOutputFormat.class, Text.class, Text.class);

		}

		TableMapReduceUtil.initTableMapperJob(job.getConfiguration().get("hbase.mapreduce.inputtable"), scan, StatsMapper.class, Text.class, ImmutableBytesWritable.class, job);

//		log.info("Table Utils" + HBaseTableUtils.listColumns());

		// [datasourceOrganization, result, organizationOrganization, person,

		// projectPerson, resultOrganization, dedup, resultResult,

		// resultProject, project, organization, personResult,

		// projectOrganization, personPerson, dedupPerson, datasource]

		SequenceFileOutputFormat.setCompressOutput(job, false);

		Path path = new Path(job.getConfiguration().get("mapred.output.dir"));

		SequenceFileOutputFormat.setOutputPath(job, path);

		return job;

	}

	private void dnetConfig(Job job) {// #HBASE-SITE

		job.getConfiguration().set("hbase.rootdir", "hdfs://nmis-hadoop-cluster/hbase");

		job.getConfiguration().set("hbase.security.authentication", "simple");

		// ZOOKEEPER

		job.getConfiguration().set("zookeeper.znode.rootserver", "root-region-server");

		job.getConfiguration().set("hbase.zookeeper.quorum",

				"quorum1.t.hadoop.research-infrastructures.eu,quorum2.t.hadoop.research-infrastructures.eu,quorum3.t.hadoop.research-infrastructures.eu,quorum4.t.hadoop.research-infrastructures.eu,jobtracker.t.hadoop.research-infrastructures.eu");

		job.getConfiguration().set("hbase.zookeeper.property.clientPort", "2182");

		// job.getConfiguration().set("dnet.clustername", "DM");

		// #CORE-SITE

		// job.getConfiguration().set("fs.defaultFS",

		// "hdfs://quorum1.t.hadoop.research-infrastructures.eu");

		//

		// job.getConfiguration().set("hadoop.security.authentication",

		// "simple");

		// job.getConfiguration().set("hadoop.security.auth_to_local",

		// "DEFAULT");

		// job.getConfiguration().set("hadoop.rpc.socket.factory.class.default",

		// "org.apache.hadoop.net.StandardSocketFactory");

		// #HDFS-SITE

		// job.getConfiguration().set("hadoop.rpc.socket.factory.class.default",

		// "org.apache.hadoop.net.StandardSocketFactory");

		// job.getConfiguration().set("dfs.ha.namenodes.nmis-hadoop-cluster",

		// "nn1,nn2 \n dfs.namenode.rpc-address.nmis-hadoop-cluster.nn1=quorum1.t.hadoop.research-infrastructures.eu:8020 \n"

		// +

		// " dfs.namenode.http-address.nmis-hadoop-cluster.nn1=quorum1.t.hadoop.research-infrastructures.eu:50070 \n "

		// +

		// " dfs.namenode.rpc-address.nmis-hadoop-cluster.nn2=quorum2.t.hadoop.research-infrastructures.eu:8020 \n "

		// +

		// "dfs.namenode.http-address.nmis-hadoop-cluster.nn2=quorum2.t.hadoop.research-infrastructures.eu:50070 \n"

		// +

		// "dfs.client.failover.proxy.provider.nmis-hadoop-cluster=org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider");

		// #MAPRED-SITE

		// job.getConfiguration().set("mapred.job.tracker", "nmis-hadoop-jt");

		// job.getConfiguration().set("mapred.jobtrackers.nmis-hadoop-jt",

		// "jt1,jt2");

		// job.getConfiguration().set("mapred.jobtracker.rpc-address.nmis-hadoop-jt.jt1",

		// "jobtracker.t.hadoop.research-infrastructures.eu:8021");

		// job.getConfiguration().set("mapred.jobtracker.rpc-address.nmis-hadoop-jt.jt2",

		// "quorum4.t.hadoop.research-infrastructures.eu:8022");

		//

		// job.getConfiguration().set("mapred.mapper.new-api", "true");

		// #OOZIE SERVER

		// job.getConfiguration().set("oozie.service.loc",

		// "http://oozie.t.hadoop.research-infrastructures.eu:11000/oozie");

	}

}