D-Net

import java.io.IOException;

import java.sql.Connection;

import java.sql.PreparedStatement;

import java.sql.Statement;

import java.util.ArrayList;

import java.util.HashSet;

import java.util.Iterator;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.io.NullWritable;

import org.apache.hadoop.io.Text;

import org.json.JSONObject;

import com.hp.hpl.jena.rdf.model.StmtIterator;

import com.jolbox.bonecp.BoneCPDataSource;

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

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 defaultGraph = "";

    private int entitiesPerQuery;

    private MapCountries mapCountries;

    private Connection conn;

    private Connection dConn;

    private JSONObject entitiesMappings;

    public static enum ENTITIES_COUNTER {

    @Override

        this.hadoopConf = context.getConfiguration();

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

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

        mapCountries = new MapCountries();

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

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

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

//            dConn = ds.getConnection();

        } 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) {

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

        while (it.hasNext()) {

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

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

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

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

            } else {

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

            }

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

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

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

        counter++;

        if (counter > entitiesPerQuery) {

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

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

//	        		Statement stmt1;

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

//	    			dConn.close();

//	                log.error("Virtuoso delete failed  at  query" + deleteQuery + " \n and with error " + e.toString(), e);

            try {

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

                ps.setString(1, insertQuery);

                ps.setString(2, "");

                final int NQUAD_LEVEL = 64;

                ps.setInt(4, NQUAD_LEVEL);

                ps.setInt(5, 0);

                ps.setInt(7, 0);

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

                ps.close();

//    			conn.close();

                buildQuery = new String[2];

//    			commitCounter+=counter;

                counter = 0;

//    			if(commitCounter>100){

//

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

            } catch (Exception e) {

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

                try {

                } catch (Exception e1) {

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

                }

        }

    }

    @Override

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

        try {

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

            ps.setString(1, insertQuery);

            ps.setString(2, "");

            final int NQUAD_LEVEL = 64;

            ps.setInt(4, NQUAD_LEVEL);

            ps.setInt(5, 0);

            ps.setInt(7, 0);

            conn.commit();

            ps.close();

            ps = conn.prepareStatement("?");

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

                conn.commit();

//            	graph.close();

//            	ds.close();

            }catch(Exception e){

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

            }

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

        }

    }

}