D-Net

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

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

import java.util.Iterator;

public class StatsReducer extends Reducer<Text, ImmutableBytesWritable, Text, Text> {

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

	private MultipleOutputs MultipleOutputWriter;

	/**

	 * 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<ImmutableBytesWritable> values, final Context context) throws IOException, InterruptedException {

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

		while (it.hasNext()) {

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

			String type = split[0];

			String id = split[1];

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

            value = value.trim();

			MultipleOutputWriter.write(type, new Text(id.getBytes(Charset.forName("UTF-8"))), new Text(value.getBytes(Charset.forName("UTF-8"))), type.toString());

		}

	}

	@Override

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

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

		MultipleOutputWriter.close();

	}

}