D-Net

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

import java.io.IOException;

import java.nio.charset.Charset;

import org.apache.commons.lang.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Reducer;

import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;

import com.google.common.base.Function;

import com.google.common.base.Joiner;

import com.google.common.collect.Iterables;

import eu.dnetlib.data.mapreduce.JobParams;

import eu.dnetlib.pace.config.DedupConfig;

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

	/**

	 * logger.

	 */

	private static final Log log = LogFactory.getLog(DedupRootsToCsvReducer.class); // NOPMD by marko on 11/24/08 5:02 PM

	private static final String COUNTER_GROUP = "csv";

	enum Tables {

		Groups, NativeGroups, NativeEntities

	}

	private DedupConfig dedupConf;

	private String DELIM;

	private String WRAP;

	private Text tKey;

	private Text tValue;

	private MultipleOutputs<Text, Text> mos;

	@Override

	protected void setup(final Context context) throws IOException, InterruptedException {

		super.setup(context);

		tKey = new Text();

		tValue = new Text();

		mos = new MultipleOutputs<Text, Text>(context);

		dedupConf = DedupConfig.load(context.getConfiguration().get(JobParams.DEDUP_CONF));

		log.info("wf conf: " + dedupConf.toString());

		DELIM = context.getConfiguration().get("mapred.textoutputformat.separator", "!");

		WRAP = context.getConfiguration().get("mapred.textoutputformat.wrapper", "#");

		log.info("using field DELIMITER: '" + DELIM + "'");

	}

	@Override

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

	InterruptedException {

		final String csv = Joiner.on(DELIM).join(Iterables.transform(values, new Function<ImmutableBytesWritable, String>() {

			@Override

			public String apply(final ImmutableBytesWritable ibw) {

				return new String(ibw.copyBytes(), Charset.forName("UTF-8"));

			}

		}));

		tKey.set(StringUtils.substringBefore(csv, DELIM));

		tValue.set(StringUtils.substringAfter(csv, DELIM));

		mos.write(Tables.Groups.toString(), tKey, tValue, Tables.Groups.toString());

		context.getCounter(COUNTER_GROUP, "groups").increment(StringUtils.countMatches(csv, DELIM));

	}

	@Override

	public void cleanup(final Context context) throws IOException, InterruptedException {

		mos.close();

	}

}