D-Net

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

import java.io.IOException;

import com.googlecode.protobuf.format.JsonFormat;

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

import eu.dnetlib.data.proto.KindProtos.Kind;

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

import eu.dnetlib.data.proto.TypeProtos.Type;

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

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Reducer;

import static eu.dnetlib.data.mapreduce.util.OafHbaseUtils.asOaf;

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

	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(final Text key, final Iterable<ImmutableBytesWritable> values, final Context context) throws IOException, InterruptedException {

		final OafRowKeyDecoder keyDecoder = OafRowKeyDecoder.decode(key.toString());

		if (!keyDecoder.getType().equals(Type.result)) {

			return;

		}

		try {

			boolean entity = false;

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

			for(Oaf oaf : asOaf(values)) {

				if (Kind.entity.equals(oaf.getKind())) {

					if (entity == true) {

						context.getCounter("result", "skipped group > 1").increment(1);

						return;

					}

					final String resulttypeid = oaf.getEntity().getResult().getMetadata().getResulttype().getClassid();

					if (!"publication".equals(resulttypeid)) {

						context.getCounter("result", "skipped " + resulttypeid).increment(1);

						return;

					}

					out.mergeFrom(oaf);

					entity = true;

				} else {

					out.getEntityBuilder().addCachedRel(oaf.getRel());

				}

			}

			outValue.set(JsonFormat.printToString(out.build()));

			context.write(outKey, outValue);

		} catch (final Exception e) {

			context.getCounter("error", e.getClass().getName()).increment(1);

			throw new RuntimeException(e);

		}

	}

}