D-Net

package eu.dnetlib.data.mapreduce.hbase.lodExport.build;

import com.lambdaworks.redis.RedisClient;

import com.lambdaworks.redis.RedisConnection;

import eu.dnetlib.data.mapreduce.hbase.lodExport.utils.Blocking;

import eu.dnetlib.data.mapreduce.hbase.lodExport.utils.LodConfiguration;

import org.apache.hadoop.fs.FileSystem;

import org.apache.hadoop.io.LongWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Mapper;

import org.apache.log4j.Logger;

import java.io.BufferedWriter;

import java.io.IOException;

import java.io.InterruptedIOException;

import java.io.OutputStream;

import java.util.HashMap;

import java.util.List;

import java.util.Map;

/*-> Parse LOD dump files

        Process lod input files and divide by entity type (both source and target)

        Transform to id, array of [ properties] format

        Store to HDFS

        For -> Multiple outputs and inputs

        Multiple inputs: all source and target datasets and their corresponding mappings

        M/O: separate output files for each dataset: mark records so that they are written to the correct one

*/

public class TargetBuildMapper extends Mapper<LongWritable, Text, Text, Text> {

    private Logger log = Logger.getLogger(this.getClass());

    private LodConfiguration lodConfiguration;

    private String lastExecutionDate;

    private RedisConnection connection;

    private RedisClient client;

    private String redisHost;

    private Integer redisPort;

    private FileSystem hdfs;

    private OutputStream os;

    private BufferedWriter br;

    private String uriPrefix;

    private String stopWords;

    private Map<String, Integer> stopWordsMap = new HashMap<>();

    private String entityType;

    public static enum TARGET_BUILD_COUNTERS {

        BLOCKING_KEYS,

        REDIS_RECORDS

    }

    @Override

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

        lodConfiguration = new LodConfiguration();

        lodConfiguration.load(context.getConfiguration().get("lod.targetMappings"));

        redisHost = context.getConfiguration().get("lod.redisHost");

        redisPort = Integer.parseInt(context.getConfiguration().get("lod.redisPort"));

        client = RedisClient.create("redis://" + redisHost + ":" + redisPort);

        log.debug("Redis connection info : " + "redis://" + redisHost + ":" + redisPort);

        connection = client.connect();

        uriPrefix = context.getConfiguration().get("lod.prefix");

        stopWords = context.getConfiguration().get("lod.stopwords");

        for (String stopword : stopWords.split(",")) {

            stopWordsMap.put(stopword, 0);

        }

        /* hdfs = FileSystem.get(context.getConfiguration());

        Path file = new Path(context.getConfiguration().get("lod.statsOutputPath") + UUID.randomUUID().toString());

        os = hdfs.create(file);

        br = new BufferedWriter(new OutputStreamWriter(os, "UTF-8"));*/

    }

    @Override

    protected void map(final LongWritable keyIn, final Text result, final Context context) throws IOException {

        try {

            //get ID

            StringBuilder id = new StringBuilder();

            String[] recordFields = result.toString().split(",");

            id.append("target_").append(recordFields[0]);

            List<String> blockingKeys = Blocking.tokenBlocking(recordFields, stopWordsMap);

            writeToRedis(id.toString(), result.toString());

            context.getCounter(TARGET_BUILD_COUNTERS.REDIS_RECORDS).increment(1);

            for (String blockingKey : blockingKeys) {

                //Write BlockingKey, RecordID to output

                context.write(new Text(blockingKey), new Text(id.toString()));

                context.getCounter(TARGET_BUILD_COUNTERS.BLOCKING_KEYS).increment(1);

            }

        } catch (Exception e) {

            log.error("Error writing entity to M/R output", e);

            log.error("result error    " + result.toString());

            throw new RuntimeException(e);

        }

    }

    private void writeToRedis(String key, String value) throws Exception {

        try {

            connection.set(key, value);

        } catch (Exception e) {

            log.error("Error writing entity to Redis", e);

            throw new RuntimeException("Error writing to Redis  ", e);

        }

    }

   /* private void writeStats(String key, String value) throws IOException {

        br.write(key + "," + value);

        br.newLine();

    }*/

    @Override

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

        super.cleanup(context);

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

    }

    public Logger getLog() {

        return log;

    }

    public void setLog(Logger log) {

        this.log = log;

    }

    public LodConfiguration getLodConfiguration() {

        return lodConfiguration;

    }

    public void setLodConfiguration(LodConfiguration lodConfiguration) {

        this.lodConfiguration = lodConfiguration;

    }

    public String getLastExecutionDate() {

        return lastExecutionDate;

    }

    public void setLastExecutionDate(String lastExecutionDate) {

        this.lastExecutionDate = lastExecutionDate;

    }

}