D-Net

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

import java.io.IOException;

import java.sql.Connection;

import java.sql.PreparedStatement;

import java.sql.SQLException;

import java.sql.Statement;

import java.util.ArrayList;

import java.util.Arrays;

import java.util.HashSet;

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;

import virtuoso.jena.driver.VirtGraph;

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 JSONObject entitiesMappings;

    private JSONObject relationsMappings;

    public static enum ENTITIES_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("entities")) 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("entities")) 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);

        }

//        this.graph = new VirtGraph(ds);

    }

    @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("entities")) {

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

            }

        }

//        System.out.println("/nID-> "+key.toString()+"\n");

//        System.out.println("/nITCOUNTER-> "+itCounter+"\n");

//        System.err.println("\nCOUNTER "+counter);

//        System.out.println("so FAR: "+buildQuery);

        counter++;

        if (counter > entitiesPerQuery) {

//            String insertQuery = "SPARQL INSERT INTO <" + defaultGraph + "> { " + buildQuery + "}";

//            Statement stmt;

////            Connection conn;

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

	        	try{

	        		String deleteQueryString ="SPARQL DELETE DATA { GRAPH <"+defaultGraph+"> {"+deleteQuery+"}}" ;

	        		Statement stmt;

	        		stmt = conn.createStatement();

	    			stmt.execute(deleteQueryString);

	    			conn.commit();

	    			stmt.close();

	    			conn.close();

	    			deleteQuery="";

	        	}catch (Exception e) {

	                log.error("Virtuoso write failed  at  query" + buildQuery[1] + " \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 failed:"+ e1.toString(), e1);

	                }

	            }

        	}

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

                ps.setInt(7, 0);

                ps.execute();

//                ps = conn.prepareStatement("commit work");

//                ps.execute();

                conn.commit();

                ps.close();

//    

//    			conn.close();

                buildQuery = new String[2];

                buildQuery[0]="";

                buildQuery[1]="";

                insertQuery="";

//    			commitCounter+=counter;

                counter = 0;

//    			if(commitCounter>100){

//    				commitCounter=0;

////    				conn.commit();

//    				conn.close();

//    				conn=ds.getConnection();

//    			}

//

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

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

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

                }

                else

                {  context.getCounter(ENTITIES_COUNTER.TOTAL_RELATIONS).increment(1);

                }

            } catch (Exception e) {

                log.error("Virtuoso write failed  at  query" + buildQuery[0] + " \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 failed:"+ e1.toString(), e1);

                }

            }

        }

    }

    @Override

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

//        String insertQuery = "SPARQL INSERT INTO <" + defaultGraph + "> { " + buildQuery + "}";

//        System.out.println("CLEAN-> NEW INSERT QUERY:  " + insertQuery);

//        Statement stmt;

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

            ps.setInt(7, 0);

            ps.execute();

            conn.commit();

            ps.close();

            ps = conn.prepareStatement("?");

            ps.setString(1,"commit work");

            ps.close();

        } catch (Exception e) {

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

//            throw new InterruptedException("Virtuoso write failed  at  query" + insertQuery + " \n and with error " + e.toString());

        } finally {

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

            try{

                conn.commit();

                conn.close();

//            	graph.close();

//            	ds.close();

            }catch(Exception e){

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

            }

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

        }

    }

}