D-Net

package eu.dnetlib.data.mapreduce.hbase.statsExport.eu.dnetlib.data.mapreduce.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 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 {

	};

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

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

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

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

		MultipleOutputWriter.close();

	}