D-Net

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

import org.apache.hadoop.hbase.io.ImmutableBytesWritable;

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.Iterator;

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

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

    private MultipleOutputs MultipleOutputWriter;

    public static enum REDUCE_COUNTER {

        WRITTEN_RECORDS

    }

    /**

     * Reducer that splits input according to their Type ( datasource, results

     * etc..) and writes each kind to a seperate output

     */

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

        while (it.hasNext()) {

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

            String[] split = value.split(context.getConfiguration().get("lod.delim"));

            value = value.replaceFirst(split[0] + context.getConfiguration().get("lod.delim"), "");

            String basePath = key.toString().equals("relations") ? "relations/" + key.toString() : "entities/" + key.toString();

            MultipleOutputWriter.write(key.toString(), new Text(split[0]), new Text(value), basePath);

            context.getCounter(REDUCE_COUNTER.WRITTEN_RECORDS).increment(1);

        }

    }

    @Override

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

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

        MultipleOutputWriter.close();

    }

}