D-Net

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

import java.io.IOException;

import java.util.Iterator;

import java.util.List;

import javax.sql.DataSource;

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 eu.dnetlib.data.mapreduce.hbase.lodImport.utils.DB;

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

import java.util.Arrays;

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

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

    private DataSource ds;

    private String fileName = "";

    private Configuration hadoopConf; 

    @Override

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

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

        this.hadoopConf = context.getConfiguration();

        try {

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

        } catch (Exception e) {

            log.error("", e);

        }

    }

    @Override

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

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

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

//		Connection conn = ds.getConnection();

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

        while (it.hasNext()) {

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

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

            System.out.println(" Values : ");

            log.info(" Values : ");

            if (fileName.startsWith("entities")) RDFizer.RDFizeEntityRow(row, ds, entitiesMappings,hadoopConf);

            else RDFizer.RDFizeRelationRow(row, ds, relationsMappings,hadoopConf);

        }

    }

    @Override

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

        log.info("Cleaning up reducer...");

    }

}