D-Net

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

import java.io.IOException;

import java.util.Arrays;

import java.util.Iterator;

import java.util.LinkedList;

import java.util.concurrent.atomic.AtomicInteger;

import com.google.common.base.Splitter;

import com.google.common.collect.Lists;

import com.google.protobuf.InvalidProtocolBufferException;

import com.googlecode.protobuf.format.JsonFormat;

import eu.dnetlib.data.mapreduce.JobParams;

import eu.dnetlib.data.mapreduce.hbase.dli.kv.DliKey;

import eu.dnetlib.data.proto.dli.ScholixObjectProtos;

import eu.dnetlib.data.proto.dli.Scholix2ObjectProtos;

import eu.dnetlib.data.transform.xml.AbstractDNetXsltFunctions;

import eu.dnetlib.dli.proto.ScholixVersion;

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;

/**

 * Created by claudio on 13/03/2017.

 */

public class PrepareScholixDataReducer extends Reducer<DliKey, ImmutableBytesWritable, Text, Text> {

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

	private Text outKey;

	private Text outValue;

	private ScholixVersion scholixVersion ;

	@Override

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

		outKey = new Text("");

		outValue = new Text();

		scholixVersion = ScholixVersion.valueOf(context.getConfiguration().get(JobParams.SCHOLIXVERSION));

	}

	@Override

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

		//System.out.println("reducing key = " + key);

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

		final ImmutableBytesWritable first = it.next();

		final AtomicInteger groupSize = new AtomicInteger(1);

		switch (scholixVersion) {

			case v1: {

				final ScholixObjectProtos.Scholix source = parsev1(key, first).build();

				if (!source.hasSource()) {

					context.getCounter("scholix", "missing source").increment(1);

					return;

				}

				it.forEachRemaining(t -> {

					groupSize.incrementAndGet();

					final ScholixObjectProtos.Scholix.Builder rel = parsev1(key, t);

					if (rel.hasTarget()) {

						rel.setSource(source.getSource());

						rel.setIdentifier(generateScholixIdentifier(rel));

						emit(context, JsonFormat.printToString(rel.build()));

					} else {

						context.getCounter("scholix", "missing target").increment(1);

					}

				});

				break;

			}

			case v2:{

				final Scholix2ObjectProtos.Scholix source = parsev2(key, first).build();

				if (!source.hasSource()) {

					context.getCounter("scholix", "missing source").increment(1);

					return;

				}

				it.forEachRemaining(t -> {

					groupSize.incrementAndGet();

					final Scholix2ObjectProtos.Scholix.Builder rel = parsev2(key, t);

					if (rel.hasTarget()) {

						rel.setSource(source.getSource());

						emit(context, JsonFormat.printToString(rel.build()));

					} else {

						context.getCounter("scholix", "missing target").increment(1);

					}

				});

				break;

			}

		}

		groupSizeCounter(context, groupSize.get(),

				"1,1",

				"1,10",

				"10,20",

				"20,100",

				"100,200",

				"200,500",

				"500,1000",

				"1000,2000",

				"2000,5000",

				"5000,10000",

				"10000,20000",

				"20000,*");

	}

	private void groupSizeCounter(final Context context, final int groupSize, final String... groups) {

		Arrays.asList(groups).forEach(g -> {

			final LinkedList<String> i = Lists.newLinkedList(Splitter.on(",").split(g));

			int min = Integer.parseInt(i.getFirst());

			int max = i.getLast().equals("*") ? Integer.MAX_VALUE : Integer.parseInt(i.getLast());

			groupSizeCounter(context, groupSize, min, max);

		});

	}

	private void groupSizeCounter(final Context context, final int groupSize, Integer min, Integer max) {

		if (groupSize > min & groupSize <= max) {

			context.getCounter("scholix groups", String.format("group size (%s,%s)", min, max)).increment(1);

		}

	}

	private ScholixObjectProtos.Scholix.Builder parsev1(DliKey key, final ImmutableBytesWritable value) {

		try {

			return ScholixObjectProtos.Scholix.newBuilder(ScholixObjectProtos.Scholix.parseFrom(value.copyBytes()));

		} catch (InvalidProtocolBufferException e) {

			throw new IllegalArgumentException(String.format("cannot parse Scholix, keytype '%s', id '%s'", key.getKeyType(), key.getId()));

		}

	}

	private String generateScholixIdentifier(ScholixObjectProtos.Scholix.Builder scholix) {

		final String relname = scholix.getRelationship().getName();

		final String sourceId = scholix.getSource().getDnetIdentifier();

		final String targetId = scholix.getTarget().getDnetIdentifier();

		return AbstractDNetXsltFunctions.md5(String.format("%s::%s::%s",sourceId,relname,targetId));

	}

	private Scholix2ObjectProtos.Scholix.Builder parsev2(DliKey key, final ImmutableBytesWritable value) {

		try {

			return Scholix2ObjectProtos.Scholix.newBuilder(Scholix2ObjectProtos.Scholix.parseFrom(value.copyBytes()));

		} catch (InvalidProtocolBufferException e) {

			throw new IllegalArgumentException(String.format("cannot parse Scholix, keytype '%s', id '%s'", key.getKeyType(), key.getId()));

		}

	}

	private void emit(final Context context, final String data) {

		outValue.set(data.getBytes());

		try {

			context.write(outKey, outValue);

		} catch (Exception e) {

			e.printStackTrace();

			throw new RuntimeException(e);

		}

	}

}