D-Net

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

import eu.dnetlib.data.mapreduce.hbase.lodExport.utils.caching.RedisUtils;

import eu.dnetlib.data.mapreduce.hbase.lodExport.utils.configuration.Properties;

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 RedisUtils redisUtils;

    private int optimalBlockSize;

    private static final char SEPERATOR = ',';

    public static enum LINK_MAPPER_COUNTERS {

        OPTIMAL_BLOCK_SIZE,

        ERROR_COUNTERS,

        DISCARDED_BLOCKS,

        PASSED_BLOCKS,

        TOTAL_COMPARISONS,

        DISCARDED_COMPARISONS,

        TOTAL_BLOCKS

    }

    @Override

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

        try {

            redisUtils = new RedisUtils(context.getConfiguration());

            optimalBlockSize = Integer.valueOf(redisUtils.getValue(Properties.OPTIMAL_BLOCKS));

            context.getCounter(LINK_MAPPER_COUNTERS.OPTIMAL_BLOCK_SIZE).setValue(optimalBlockSize);

            log.info("optimalBlockSize " + optimalBlockSize);

        } catch (Exception e) {

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

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

        }

    }

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

            //purge blocks that contain only source or target entities

            //how many comparisons we have purged

            context.getCounter(LINK_MAPPER_COUNTERS.TOTAL_COMPARISONS).increment(recordsNumber * (recordsNumber - 1));

            if (recordsNumber >= optimalBlockSize) {

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

                //TODO count source / target records and get actual comparison number

                context.getCounter(LINK_MAPPER_COUNTERS.DISCARDED_COMPARISONS).increment(recordsNumber * (recordsNumber - 1));

            } else {

                context.write(keyIn, result);

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

            }

            context.getCounter(LINK_MAPPER_COUNTERS.TOTAL_BLOCKS).increment(1);

        } catch (Exception e) {

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

            log.error("ERROR ", e);

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

            throw new IOException(e.toString(), e);

        }

    }

    private static int countRecords(Text block) {

        int count = 0;

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

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

                count++;

            }

        }

        return count;

    }

}