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.hbase.dli.kv.DliKey;

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

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;

	@Override

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

		outKey = new Text("");

		outValue = new Text();

	}

	@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 Scholix source = parse(key, first).build();

		final AtomicInteger groupSize = new AtomicInteger(1);

		if (!source.hasSource()) {

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

			return;

		}

		it.forEachRemaining(t -> {

			groupSize.incrementAndGet();

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

			if (rel.hasTarget()) {

				rel.setSource(source.getSource());

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

			} else {

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

			}

		});

		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 Scholix.Builder parse(DliKey key, final ImmutableBytesWritable value) {

		try {

			return Scholix.newBuilder(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);

		}

	}

}