D-Net

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

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.nio.charset.Charset;

import java.util.Iterator;

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

    public static enum ENTITIES_COUNTER {

        TOTAL_ENTITIES,

    }

    private Logger log = Logger.getLogger(DatasetReducer.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();

        StringBuffer value = new StringBuffer();

        String[] tmp = key.toString().split(",");

        String dataset = tmp[0];

        String entityType = tmp[1];

        String id = tmp[2];

        while (it.hasNext()) {

            String field = new String(it.next().copyBytes(), Charset.forName("UTF-8"));

            value.append(field);

        }

        try {

            String output =entityType;

            String outputFile = dataset + "/" + output;

            MultipleOutputWriter.write(output, new Text("id\t" + id),

                    new Text(value.toString().getBytes(Charset.forName("UTF-8"))), outputFile);

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

        } catch (Exception e) {

            log.error(entityType + e.toString());

            throw new InterruptedException(entityType + " " + id + " " + e.toString());

        }

    }

    @Override

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

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

        MultipleOutputWriter.close();

    }

}