D-Net

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

import java.io.IOException;

import java.nio.charset.Charset;

import java.util.Iterator;

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;

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

    private static final String SEPERATOR = ",";

    public static enum BLOCKS_COUNTER {

        TOTAL_ENTITIES

    }

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

    private MultipleOutputs MultipleOutputWriter;

    @Override

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

        MultipleOutputWriter = new MultipleOutputs((TaskInputOutputContext) context);

    }

    @Override

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

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

        try {

            //each list is a block

            StringBuilder field = new StringBuilder();

            int nunberOfEntities = 0;

            while (it.hasNext()) {

                field.append(it.next()).append(SEPERATOR);

                nunberOfEntities++;

            }

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

            MultipleOutputWriter.write("entitiesNumber", key, new Text(String.valueOf(nunberOfEntities).getBytes(Charset.forName("UTF-8"))), "stats/entitiesNumber");

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

        } catch (Exception e) {

            throw new InterruptedException(e.getMessage());

        }

    }

    @Override

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

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

        MultipleOutputWriter.close();

    }

}