D-Net

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

import java.io.IOException;

import java.util.Iterator;

import com.google.protobuf.InvalidProtocolBufferException;

import com.googlecode.protobuf.format.JsonFormat;

import eu.dnetlib.data.mapreduce.hbase.dli.key.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.io.Text;

import org.apache.hadoop.mapreduce.Reducer;

/**

 * Created by claudio on 13/03/2017.

 */

public class PrepareScholixDataReducer extends Reducer<DliKey, Text, 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<Text> values, Context context) throws IOException, InterruptedException {

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

		final Scholix source = parse(it.next()).build();

		if (source.hasSource() & !source.hasTarget()) {

			throw new IllegalStateException("sorting does not work! Expected entity as 1st element");

		}

		it.forEachRemaining(t -> {

			final Scholix.Builder rel = parse(t);

			rel.setSource(source.getSource());

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

		});

	}

	private Scholix.Builder parse(final Text t) {

		try {

			return Scholix.newBuilder(Scholix.parseFrom(t.getBytes()));

		} catch (InvalidProtocolBufferException e) {

			throw new IllegalArgumentException("cannot parse Scholix: " + t.toString());

		}

	}

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

		}

	}

}