D-Net

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

import java.io.IOException;

import java.util.Iterator;

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;

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

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

	private MultipleOutputs MultipleOutputWriter;

	private long counter = 0;

	/**

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

			counter++;

			value = value.trim();

			MultipleOutputWriter.write(type, "	" + new Text(id), new Text(context.getConfiguration().get("stats.delim") + value), type.toString());

		}

	}

	@Override

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

		log.info("******************************* TOTAL RECORDS  " + counter);

		MultipleOutputWriter.close();

	}

}