D-Net

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

import com.lambdaworks.redis.RedisClient;

import com.lambdaworks.redis.RedisConnection;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Mapper;

import org.apache.log4j.Logger;

import java.io.IOException;

import java.io.InterruptedIOException;

public class LinkMapper extends Mapper<Text, Text, Text, Text> {

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

    private String redisHost;

    private Integer redisPort;

    private RedisConnection connection;

    private RedisClient client;

    private int optimalBlockSize;

    private static final char SEPERATOR = ',';

    private static final String OPTIMAL_BLOCKS = "optimalBlockSize";

    public static enum TEST_COUNTERS {

        OPTIMAL_BLOCK_SIZE,

        ERROR_COUNTERS,

        DISCARDED_BLOCKS,

        PASSED_BLOCKS,

        TOTAL_COMPARISONS,

        DISCARDED_COMPARISONS

    }

    @Override

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

        try {

            initRedis(context);

            optimalBlockSize = Integer.valueOf((String) connection.get(OPTIMAL_BLOCKS));

            //optimalBlockSize=90000;

            context.getCounter(TEST_COUNTERS.OPTIMAL_BLOCK_SIZE).setValue(Long.parseLong((String) connection.get("optimalBlockSize")));

        } catch (Exception e) {

            log.error("ERROR CONNECTING TO REDIS ", e);

            throw new InterruptedIOException("ERROR CONNECTING TO REDIS " + e.toString());

        }

    }

    private void initRedis(Context context) {

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

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

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

        connection = client.connect();

    }

    @Override

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

        try {

            //purge blocks with number of  records > optimal

            int recordsNumber = countRecords(result, SEPERATOR);

            //purge blocks that contain only source or target entities

            //          boolean hasBothSouceAndTarget = resultString.contains("source_") && result.toString().contains("target_");

            //how many comparisons we have purged

            context.getCounter(TEST_COUNTERS.TOTAL_COMPARISONS).increment(recordsNumber * recordsNumber);

            //       if (recordsNumber == 1 || recordsNumber >= optimalBlockSize || !hasBothSouceAndTarget) {

            if (recordsNumber >= optimalBlockSize) {

                context.getCounter(TEST_COUNTERS.DISCARDED_BLOCKS).increment(1);

                context.getCounter(TEST_COUNTERS.DISCARDED_COMPARISONS).increment(recordsNumber * recordsNumber);

            } else {

                context.write(keyIn, result);

                context.getCounter(TEST_COUNTERS.PASSED_BLOCKS).increment(1);

            }

        } catch (Exception e) {

            log.error("key causing error is " + keyIn);

            log.error("ERROR ", e);

            context.getCounter(TEST_COUNTERS.ERROR_COUNTERS).increment(1);

            throw new InterruptedIOException(e.toString());

        }

    }

    private static int countRecords(Text block, char seperator) {

        int count = 0;

        for (int i = 0; i < block.getLength(); i++) {

            if (block.charAt(i) == seperator) {

                count++;

            }

        }

        return count;

    }

    @Override

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

        if (connection != null) {

            connection.close();

        }

        super.cleanup(context);

    }

}