D-Net

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

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

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Reducer;

import org.apache.hadoop.mapreduce.TaskInputOutputContext;

import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;

import org.apache.log4j.Logger;

import java.io.IOException;

import java.util.HashMap;

import java.util.Iterator;

import java.util.Map;

;

public class BlockNoCacheReducer extends Reducer<Text, Text, Text, Text> {

    private static final String SEPERATOR = ",";

    private static final String FIELD_DELIM = "\t";

    private static String ERROR_KEY = "";

    private static int RECORD_LIMIT = 10000;

    public static enum BLOCKS_COUNTER {

        BLOCKS,

        DISCARDED_BLOCKS,

        REDIS_RECORDS

    }

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

    private MultipleOutputs MultipleOutputWriter;

    private RedisUtils redisUtils;

    @Override

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

        MultipleOutputWriter = new MultipleOutputs((TaskInputOutputContext) context);

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

    }

    @Override

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

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

        StringBuilder field = new StringBuilder();

        Map<String, String> records = new HashMap<>();

        try {

            boolean hasSource = false;

            boolean hasTarget = false;

            int entitiesNumber = 0;

            while (it.hasNext()) {

                entitiesNumber++;

                if (entitiesNumber >= RECORD_LIMIT) {

                    break;

                }

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

                String id = record.substring(0, record.indexOf(FIELD_DELIM));

                field.append(id).append(SEPERATOR);

                if (id.contains("source_")) {

                    hasSource = true;

                }

                if (id.contains("target_")) {

                    hasTarget = true;

                }

                records.put(id, record);

            }

            if (hasSource && hasTarget) {

                redisUtils.writeMap(records);

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

            }

            if (hasSource && hasTarget && entitiesNumber < RECORD_LIMIT) {

                MultipleOutputWriter.write("blocks", key, field.toString(), "blocks/b");

                MultipleOutputWriter.write("entitiesNumber", key, new Text(String.valueOf(entitiesNumber)), "stats/entitiesNumber");

                context.getCounter(BLOCKS_COUNTER.BLOCKS).increment(1);

            } else {

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

            }

        } catch (Exception e) {

            ERROR_KEY = key.toString();

            log.error("key that caused the issue: " + ERROR_KEY);

            throw new IOException("Error writing out entities. key that caused the issue: " + ERROR_KEY + "", e);

        }

    }

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

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

        if (!ERROR_KEY.isEmpty()) {

            log.error("key that caused the issue: " + ERROR_KEY);

        }

        redisUtils.close();

    }

}