D-Net

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

import java.io.IOException;

import java.util.List;

import org.apache.hadoop.io.LongWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Mapper;

import org.apache.log4j.Logger;

import org.joda.time.DateTime;

import org.joda.time.format.DateTimeFormat;

import org.joda.time.format.DateTimeFormatter;

import eu.dnetlib.data.mapreduce.hbase.index.config.EntityConfigTable;

import eu.dnetlib.data.mapreduce.hbase.index.config.IndexConfig;

public class LodImportMapper extends Mapper<LongWritable, List<Text>, Text, List<Text>> {

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

	private EntityConfigTable entityConfigTable;

	private long threshold = 100000;

	private String fileName = "";

	private String lastExecutionDate = "";

	@Override

	protected void setup(Mapper.Context context) throws IOException, InterruptedException {

		loadEntityConfig(context);

		this.fileName = context.getConfiguration().get("lod.inputFile");

		this.lastExecutionDate = context.getConfiguration().get("lod.lastExecutionDate");

	}

	@Override

	protected void map(LongWritable key, List<Text> values, Context context) throws IOException, InterruptedException {

		DateTimeFormatter formatter = DateTimeFormat.forPattern("yyyy/MM/dd HH:mm:ss");

		String dateOfTransformationString = values.get(2).toString();

		DateTime dt = formatter.parseDateTime(dateOfTransformationString);

		DateTime de = formatter.parseDateTime(lastExecutionDate);

		Text keyString = values.get(1);

		//TODO changed output format form String, List<String> to Text,List<Text>

 		if (lastExecutionDate == "" || lastExecutionDate == null || dateOfTransformationString == "") context.write(keyString, values);

		if (dt.isBefore(de)) context.write(keyString, values);

	}

	private void loadEntityConfig(Context context) {

		String indexConf = context.getConfiguration().get("lod.indexConf");

		if (indexConf == null || indexConf.isEmpty()) {

			log.error("NULL ENTITY CONFIGURATION TABLE IN MAPPER  : ");

		}

		entityConfigTable = IndexConfig.load(indexConf).getConfigMap();

	}

	@Override

	protected void cleanup(Context context) throws IOException, InterruptedException {

		super.cleanup(context);

	}

	public EntityConfigTable getEntityConfigTable() {

		return entityConfigTable;

	}

	public void setEntityConfigTable(EntityConfigTable entityConfigTable) {

		this.entityConfigTable = entityConfigTable;

	}

	public long getThreshold() {

		return threshold;

	}

	public void setThreshold(long threshold) {

		this.threshold = threshold;

	}

}