D-Net

import java.io.IOException;

import java.sql.Connection;

import java.sql.PreparedStatement;

import java.util.ArrayList;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.io.NullWritable;

import org.apache.hadoop.io.Text;

import org.json.JSONObject;

import com.jolbox.bonecp.BoneCPDataSource;

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

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 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 REDUCES_COUNTER {

        RESULT,

    @Override

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

        mapCountries = new MapCountries();

        buildQuery[0] = "";

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

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

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

            } else {

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

                insertQuery += buildQuery[0];

            }

        counter++;

        if (counter > entitiesPerQuery ) {

                 ps.setInt(6, 15);

                 ps.setInt(7, 0);

                 ps.close();

                 buildQuery = new String[2];

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

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

                 } else {

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

                 }

        }catch(Exception e){

            try {

                if (conn != null) {

                } else {

                    conn.close();

                }

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

            } catch (Exception e1) {

            }

          }

        }

    }

    @Override

        try {

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

            ps.setString(1, insertQuery);

            ps.setString(2, "");

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

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

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

            } else {

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

            }

        } catch (Exception e) {

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

                    ps.setInt(6, 15);

                    ps.setInt(7, 0);

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

                    } else {

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

                    }

                    ps.setInt(6, 15);

                    ps.setInt(7, 0);

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

                    } else {

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

                    }

        } finally {

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

            try {

                conn.commit();

            } catch (Exception e) {

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

            }

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

        }

    }

}