D-Net

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

import java.io.IOException;

import java.sql.Connection;

import java.sql.PreparedStatement;

import java.util.ArrayList;

import java.util.Arrays;

import java.util.Iterator;

import java.util.List;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.io.NullWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Reducer;

import org.apache.log4j.Logger;

import org.json.JSONObject;

import com.jolbox.bonecp.BoneCPDataSource;

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

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

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

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

public class LodImportReducer extends Reducer<Text, Text, NullWritable, NullWritable> {

    private Logger log = Logger.getLogger(LodImportReducer.class);

    private BoneCPDataSource ds;

    private String fileName = "";

    private Configuration hadoopConf;

    //    private VirtGraph graph;

    private int counter;

    private int commitCounter;

    private String[] buildQuery = new String[2];

    private String insertQuery = "";

    private String deleteQuery = "";

    private String defaultGraph = "";

    private int entitiesPerQuery;

    private MapCountries mapCountries;

    private MapLanguages mapLanguages;

    private Connection conn;

    private Connection dConn;

    private JSONObject entitiesMappings;

    private JSONObject relationsMappings;

    public static enum REDUCES_COUNTER {

        RESULT,

        PROJECT,

        DATASOURCE,

        PERSON,

        ORGANIZATION,

        TOTAL_ENTITIES,

        TOTAL_RELATIONS

    }

    @Override

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

        this.hadoopConf = context.getConfiguration();

        this.fileName = hadoopConf.get("lod.inputFile");

        if (!fileName.contains("relations")) defaultGraph = hadoopConf.get("lod.defaultGraph");

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

        this.counter = 0;

        mapCountries = new MapCountries();

        mapLanguages = new MapLanguages();

        buildQuery[0] = "";

        buildQuery[1] = "";

        entitiesMappings = new JSONObject(hadoopConf.get("lod.jsonEntities"));

        relationsMappings = new JSONObject(hadoopConf.get("lod.jsonRels"));

        if (!fileName.contains("relations")) entitiesPerQuery = Integer.parseInt(hadoopConf.get("lod.entitiesPerQuery"));

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

        try {

            DB db = new DB();

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

            conn = ds.getConnection();

            conn.setAutoCommit(false);

        } catch (Exception e) {

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

        }

    }

    @Override

    protected void reduce(final Text key, final Iterable<Text> values, final Context context) throws IOException, InterruptedException {

        if (entitiesMappings == null || relationsMappings == null) {

            throw new InterruptedException("Could not read json mappings!");

        }

        Iterator<Text> it = values.iterator();

        while (it.hasNext()) {

            String data = it.next().toString();

            String[] split = data.split(hadoopConf.get("lod.delim"));

            List<String> row = new ArrayList<String>(Arrays.asList(split));

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

                buildQuery = RDFizer.RDFizeEntityRow(row, ds, entitiesMappings, hadoopConf, mapCountries, mapLanguages, defaultGraph);

                insertQuery += buildQuery[0];

                deleteQuery += buildQuery[1];

            } else {

                buildQuery = RDFizer.RDFizeRelationRow(row, relationsMappings, hadoopConf);

                insertQuery += buildQuery[0];

            }

        }

        counter++;

        if (counter > entitiesPerQuery ) {

        	 try {

                 PreparedStatement ps = conn.prepareStatement("DB.DBA.TTLP_MT (?, ?, ?, ?, ?, ?,?)");

                 ps.setString(1, insertQuery);

                 ps.setString(2, "");

                 ps.setString(3, defaultGraph);

                 final int NQUAD_LEVEL = 64;

                 ps.setInt(4, NQUAD_LEVEL);

                 ps.setInt(5, 0);

                 ps.setInt(6, 12);

                 ps.setInt(7, 0);

                 ps.execute();                 

                 conn.commit();

                 ps.close();

                 buildQuery = new String[2];

                 buildQuery[0] = "";

                 buildQuery[1] = "";

                 insertQuery = "";

                 counter = 0;

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

                     //s// TODO: keep counters for all entity types

                	 context.getCounter(REDUCES_COUNTER.TOTAL_ENTITIES).increment(entitiesPerQuery);    

                 } else {

                     context.getCounter(REDUCES_COUNTER.TOTAL_RELATIONS).increment(entitiesPerQuery);

                 }   	

        }catch(Exception e){

        	log.error("Virtuoso write failed  at  query: " + insertQuery + " \n and with error " + e.toString(), e);

            try {

                if (conn != null) {

                    conn.rollback();

                    conn.close();

                    conn = ds.getConnection();

                    conn.setAutoCommit(false);

                } else {

                    conn.close();

                    conn = ds.getConnection();

                    conn.setAutoCommit(false);

                }

            } catch (Exception e1) {

                log.error("Getting another Connection fail:" + e1.toString(), e1);

            }

          }

        }

    }

    @Override

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

        try {

        	log.info("########### \n Cleanup Started. Commiting last Triples...\n##########");

        	PreparedStatement ps = conn.prepareStatement("DB.DBA.TTLP_MT (?, ?, ?, ?, ?, ?,?)");

            ps.setString(1, insertQuery);

            ps.setString(2, "");

            ps.setString(3, defaultGraph);

            final int NQUAD_LEVEL = 64;

            ps.setInt(4, NQUAD_LEVEL);

            ps.setInt(5, 0);

            ps.setInt(6, 12);

            ps.setInt(7, 0);

            ps.execute();                 

            conn.commit();

            ps.close();

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

                //s// TODO: keep counters for all entity types

           	 context.getCounter(REDUCES_COUNTER.TOTAL_ENTITIES).increment(entitiesPerQuery);    

            } else {

                context.getCounter(REDUCES_COUNTER.TOTAL_RELATIONS).increment(entitiesPerQuery);

            }   	

        } catch (Exception e) {

        	log.error("Virtuoso write failed  at  query: " + insertQuery + " \n and with error " + e.toString(), e);

            try {

                if (conn != null) {

                    conn.rollback();

                    conn.close();

                    conn = ds.getConnection();

                    conn.setAutoCommit(false);

                    PreparedStatement ps = conn.prepareStatement("DB.DBA.TTLP_MT (?, ?, ?, ?, ?, ?,?)");

                    ps.setString(1, insertQuery);

                    ps.setString(2, "");

                    ps.setString(3, defaultGraph);

                    final int NQUAD_LEVEL = 64;

                    ps.setInt(4, NQUAD_LEVEL);

                    ps.setInt(5, 0);

                    ps.setInt(6, 12);

                    ps.setInt(7, 0);

                    ps.execute();                 

                    conn.commit();

                    ps.close();

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

                        //s// TODO: keep counters for all entity types

                   	 context.getCounter(REDUCES_COUNTER.TOTAL_ENTITIES).increment(entitiesPerQuery);    

                    } else {

                        context.getCounter(REDUCES_COUNTER.TOTAL_RELATIONS).increment(entitiesPerQuery);

                    }   	

                } else {

                    conn.close();

                    conn = ds.getConnection();

                    conn.setAutoCommit(false);

                    PreparedStatement ps = conn.prepareStatement("DB.DBA.TTLP_MT (?, ?, ?, ?, ?, ?,?)");

                    ps.setString(1, insertQuery);

                    ps.setString(2, "");

                    ps.setString(3, defaultGraph);

                    final int NQUAD_LEVEL = 64;

                    ps.setInt(4, NQUAD_LEVEL);

                    ps.setInt(5, 0);

                    ps.setInt(6, 12);

                    ps.setInt(7, 0);

                    ps.execute();                 

                    conn.commit();

                    ps.close();

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

                        //s// TODO: keep counters for all entity types

                   	 context.getCounter(REDUCES_COUNTER.TOTAL_ENTITIES).increment(entitiesPerQuery);    

                    } else {

                        context.getCounter(REDUCES_COUNTER.TOTAL_RELATIONS).increment(entitiesPerQuery);

                    }   	

                }

            } catch (Exception e1) {

                log.error("Getting another Connection fail:" + e1.toString(), e1);

            }

        } finally {

            log.info("Cleaning up reducer...\nClosing Graph and Datasource");

            try {

                conn.commit();

                conn.close();

            } catch (Exception e) {

                log.error("Could not Close connection \n" + e.toString(), e);

            }

            log.info("Graph and Datasource Closed...");

        }

    }

}