D-Net

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

import java.io.IOException;

import java.nio.charset.Charset;

import java.util.List;

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.googlecode.protobuf.format.JsonFormat;

import com.googlecode.protobuf.format.JsonFormat.ParseException;

import eu.dnetlib.data.mapreduce.JobParams;

import eu.dnetlib.data.mapreduce.util.OafDecoder;

import eu.dnetlib.data.proto.OafProtos.Oaf;

import eu.dnetlib.pace.util.DedupConfig;

import eu.dnetlib.pace.util.DedupConfigLoader;

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 = DedupConfigLoader.load(context.getConfiguration().get(JobParams.WF_CONF));

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

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

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

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

	}

	@Override

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

	InterruptedException {

		for (final ImmutableBytesWritable ibw : values) {

			final RootEntity rootEntity = RootEntity.decode(new String(ibw.copyBytes(), Charset.forName("UTF-8")));

			emitRoot(context, rootEntity.getJoaf());

			emitRels(context, key, rootEntity.getMergedIds());

			return;

		}

	}

	private void emitRoot(final Context context, final String joaf) throws IOException, InterruptedException {

		final Oaf.Builder builder = Oaf.newBuilder();

		try {

			JsonFormat.merge(joaf, builder);

		} catch (final ParseException e) {

			context.getCounter("roots csv", e.getClass().getSimpleName()).increment(1);

			return;

		}

		final OafDecoder d = OafDecoder.decode(builder.build());

		tKey.set((WRAP + d.getEntityId() + WRAP).getBytes(Charset.forName("UTF-8")));

		// tValue.set(value.getBytes(Charset.forName("UTF-8")));

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

		context.getCounter(COUNTER_GROUP, "native_entities").increment(1);

	}

	private void emitRels(final Context context, final ImmutableBytesWritable key, final List<String> mergedIds) throws IOException, InterruptedException {

		final StringBuilder sb = new StringBuilder();

		final String groupId = new String(key.copyBytes());

		// native_groups groups native_entities

		tKey.set((WRAP + groupId + WRAP).getBytes(Charset.forName("UTF-8")));

		tValue.set((WRAP + dedupConf.getConfigurationId() + WRAP).getBytes(Charset.forName("UTF-8")));

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

		context.getCounter(COUNTER_GROUP, "groups").increment(mergedIds.size());

		for (final String id : mergedIds) {

			sb.append(WRAP).append(id).append(WRAP).append(DELIM);

			tValue.set(sb.toString().getBytes(Charset.forName("UTF-8")));

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

		}

		context.getCounter(COUNTER_GROUP, "native_groups").increment(mergedIds.size());

	}

	@Override

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

		mos.close();

	}

}