D-Net

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

import com.google.common.collect.ArrayListMultimap;

import com.google.common.collect.Multimap;

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

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

import org.apache.hadoop.io.LongWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Mapper;

import org.apache.log4j.Logger;

import redis.clients.jedis.Jedis;

import redis.clients.jedis.JedisPool;

import redis.clients.jedis.exceptions.JedisException;

import java.io.IOException;

import java.util.ArrayList;

import java.util.List;

import java.util.Map;

/**

 * Mapper Class that reads HBASE contents and prepares them for the StatsDB

 * export

 */

/*

-> 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 SourceBuildMapper extends Mapper<LongWritable, Text, Text, Text> {

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

    //address of your redis server

    private String redisHost;

    private Integer redisPort;

    //the jedis connection pool..

    private static JedisPool pool = null;

    private LodConfiguration lodConfiguration;

    private String lastExecutionDate;

    public static enum ENTITIES_COUNTER {

        BLOCKING_KEYS,

        REDIS_RECORDS

    }

    @Override

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

        lodConfiguration = new LodConfiguration();

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

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

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

        pool = new JedisPool(redisHost, redisPort);

    }

    @Override

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

        try {

            //get ID - output source_recordID so we can group by id and get all props of a record

            StringBuffer value = new StringBuffer();

            Multimap<String, String> Fields = ArrayListMultimap.create();

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

            for (String pair : pairs) {

                String[] split = pair.split("\t");

                        System.out.println(split[0] + split[1]);

                if (split.length > 1) {

                    Fields.put(split[0], split[1]);

                } else {

                    Fields.put(split[0], " ");

                }

            }

            /*String entityType = Fields.get("http://www.w3.org/1999/02/22-rdf-syntax-ns#type,").toString()

                    .substring(keyIn.toString().lastIndexOf("/"), keyIn.toString().length());

*/

            //get ID

            String id = "source_" + (String) ((List) Fields.get("id")).get(0);

            for (Map.Entry<String, String> entry : Fields.entries()) {

                //   if (lodConfiguration.isValidField(entityType, entry.getKey().replace("/", "\\/")))

                {

                    value.append(entry.getKey() + "\t" + entry.getValue() + ",");

                }

            }

            //Write to cache

            writeToRedis(id, value.toString());

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

            ArrayList<String> blockingKeys = Blocking.tokenBlocking(Fields);

            for (String blockingKey : blockingKeys) {

                //Write BlockingKey, RecordID to output

                System.out.println("Writing out " + blockingKey + " , " + id);

                log.info("Writing out " + blockingKey + " , " + id);

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

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

            }

        } catch (Exception e) {

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

            throw new RuntimeException(e);

        }

    }

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

        //get a jedis connection jedis connection pool

        Jedis jedis = pool.getResource();

        try {

            //save to redis

            jedis.sadd(key, value);

            //here check

            //Set members = jedis.smembers(key);

            //  jedis.hmset(key, Fields.asMap().);

            //after saving the data, lets retrieve them to be sure that it has really added in redis

           /*     Map<String, String> retrieveMap = jedis.hgetAll(key);

                for (String keyMap : retrieveMap.keySet()) {

                    System.out.println(keyMap + " " + retrieveMap.get(keyMap));

                }*/

        } catch (JedisException e) {

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

            if (null != jedis) {

                pool.returnBrokenResource(jedis);

                jedis = null;

            }

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

        } finally

        {

            if (null != jedis) pool.returnResource(jedis);

        }

    }

    @Override

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

        super.cleanup(context);

    }

    public Logger getLog() {

        return log;

    }

    public void setLog(Logger log) {

        this.log = log;

    }

    public String getRedisHost() {

        return redisHost;

    }

    public void setRedisHost(String redisHost) {

        this.redisHost = redisHost;

    }

    public Integer getRedisPort() {

        return redisPort;

    }

    public void setRedisPort(Integer redisPort) {

        this.redisPort = redisPort;

    }

    public static JedisPool getPool() {

        return pool;

    }

    public static void setPool(JedisPool pool) {

        SourceBuildMapper.pool = pool;

    }

    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;

    }

}