D-Net

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

import java.io.IOException;

import java.sql.Connection;

import java.sql.SQLException;

import java.sql.Statement;

import org.apache.hadoop.conf.Configuration;

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 com.jolbox.bonecp.BoneCPDataSource;

import eu.dnetlib.data.mapreduce.hbase.lodImport.utils.DB;

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

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

    private String fileName = "";

    private String lastExecutionDate = "";

    private Configuration hadoopConf;

    private BoneCPDataSource ds;

    private Connection conn;

    String baseURI;

    private String defaultGraph = "";

    private String deleteQuery = "";

    private int counter;

    private int entitiesPerQuery;

    public static enum ENTITIES_COUNTER {

        STORED, DISCARDED,

        ENTITIES,RELATIONS,NULL_T_DATES,DELETED

    }

    @Override

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

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

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

        this.hadoopConf = context.getConfiguration();

        if (fileName.contains("entities")){

        	defaultGraph = hadoopConf.get("lod.defaultGraph");

        	entitiesPerQuery = Integer.parseInt(hadoopConf.get("lod.entitiesPerQuery"));

        }

        else{

        	defaultGraph = hadoopConf.get("lod.relationsGraph");

        	entitiesPerQuery = Integer.parseInt(hadoopConf.get("lod.relationsPerQuery"));

        }

        baseURI = hadoopConf.get("lod.baseURI");

        try {

            DB db = new DB();

            ds = db.getDatasource(hadoopConf.get("lod.conLine"), hadoopConf.get("lod.username"), hadoopConf.get("lod.password"), "1", "1", "1");

            conn = ds.getConnection();

            conn.setAutoCommit(false);

            Statement stmt1;

    		stmt1 = conn.createStatement();

    		stmt1.execute("__dbf_set ('enable_qp', 1)");

    		stmt1.close();

        } catch (Exception e) {

            log.error(e.toString(), e);

        }

    }

    @Override

    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

        DateTimeFormatter formatter = DateTimeFormat.forPattern("yyyy-MM-dd");

        //TODO remove enclosing  delim from csv -not needed

        String split[] = value.toString().split(context.getConfiguration().get("lod.delim"));

        if (split.length > 1) {

//            Text outputKey = new Text(split[1]);

//            Text outputValue = new Text();

            Text outputKey = null;

            Text outputValue = null;

            String resourceURI="";

            try {

                outputKey = new Text(split[1]);

                outputValue = new Text(value.toString().replace("\\"," "));

                resourceURI = baseURI+split[0]+"/"+split[1];

            } catch (Exception e) {

                log.error(split.toString(), e);

            }

//		String insertQuery = "INSERT INTO <debug> { <http://test.org/"+outputKey.toString().trim()+"> a foaf:Person; <http://test.prop/1> \""+cleanValue+"\"}";

//		VirtuosoUpdateRequest q2ur = VirtuosoUpdateFactory.create(insertQuery, graph);

//		q2ur.exec();

            if (fileName.contains("relations")) {

                outputKey = new Text(key.toString());

                writeOut(context, outputKey, outputValue);

                context.getCounter(ENTITIES_COUNTER.RELATIONS).increment(1);

            } else {

                context.getCounter(ENTITIES_COUNTER.ENTITIES).increment(1);

                String dateOfTransformationString = split[2].trim();

//                if (dateOfTransformationString == null || dateOfTransformationString.isEmpty() || dateOfTransformationString.equals("")){

//                	writeOut(context, outputKey, outputValue); 

//                	context.getCounter(ENTITIES_COUNTER.STORED).increment(1);

                	 context.getCounter(ENTITIES_COUNTER.NULL_T_DATES).increment(1);

//                	

//                	String deleteQuery ="<"+resourceURI+"> ?p ?o. ";

//                	delete(deleteQuery);

//                	}

                if (dateOfTransformationString.contains("T")) {

                    int index = dateOfTransformationString.indexOf("T");

                    dateOfTransformationString = dateOfTransformationString.substring(0, index);

                }

                if (lastExecutionDate == "" || lastExecutionDate == null || dateOfTransformationString == null || dateOfTransformationString == "" || dateOfTransformationString.equals("null") || dateOfTransformationString.isEmpty() || dateOfTransformationString.equals("")) {

//                    writeOut(context, outputKey, outputValue);

                    context.getCounter(ENTITIES_COUNTER.NULL_T_DATES).increment(1);

//                    context.getCounter(ENTITIES_COUNTER.STORED).increment(1);

//                    String deleteString ="<"+resourceURI+"> ?p ?o. ";

//                	deleteQuery+=deleteString;

                } else {

//               System.err.println("Transformation:\t"+dateOfTransformationString+" \t"+outputKey+" \t"+dateOfTransformationString.length());

                    DateTime dt = formatter.parseDateTime(dateOfTransformationString);

                    DateTime de = formatter.parseDateTime(lastExecutionDate);

                    if (de.isBefore(dt)) {

                    	writeOut(context, outputKey, outputValue); 

                    	context.getCounter(ENTITIES_COUNTER.STORED).increment(1);

                    	String deleteString ="<"+resourceURI+"> ?p ?o. ";

                    	deleteQuery+=deleteString;

                    	}

                    else { context.getCounter(ENTITIES_COUNTER.DISCARDED).increment(1);}

                }

                counter++;;

                if (counter > entitiesPerQuery) {

                	delete(deleteQuery);

                	deleteQuery="";

                	counter = 0;

                	context.getCounter(ENTITIES_COUNTER.DELETED).increment(entitiesPerQuery);

                }

            }

        }

    }

    @Override

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

        super.cleanup(context);

        delete(deleteQuery);

        try {

			conn.close();

		} catch (SQLException e) {

			log.error(e.toString(), e);

		}

    }

    private void writeOut(Context context, Text outputKey, Text outputValue) throws IOException, InterruptedException {

//        log.info("Written out "+  outputKey);

        context.write(outputKey, outputValue);

    }

    private void delete(String deleteQuery){

    	String deleteQueryString ="SPARQL DEFINE sql:log-enable 0 WITH <"+defaultGraph+"> DELETE {"+deleteQuery+"} WHERE {"+deleteQuery+"}" ;

		try{

	    	Statement stmt;

			stmt = conn.createStatement();

			stmt.execute(deleteQueryString);

			conn.commit();

			stmt.close();

		}catch (Exception e){

			 log.error(e.toString(), e);

		}

    }

}