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.hadoopConf = context.getConfiguration();

        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();

            conn.close();

        } catch (Exception e) {

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

        }

    }

    @Override

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

        //TODO remove enclosing  delim from csv -not needed

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

        if (split.length > 1) {

            Text outputKey = null;

            Text outputValue = null;

            try {

                outputKey = new Text(split[1]);

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

            } catch (Exception e) {

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

            }

            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);

                writeOut(context, outputKey, outputValue);

            }

        }

    }

    @Override

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

        super.cleanup(context);

        if(conn!=null){

	        try {

	            conn.close();

	        } catch (SQLException e) {

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

	        }

        }

    }

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

        context.write(outputKey, outputValue);

    }

}