D-Net

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

import java.io.IOException;

import java.util.List;

import java.util.Objects;

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 eu.dnetlib.data.mapreduce.hbase.broker.model.EventWrapper;

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

	private static final int LIMIT = 1000;

	private Text tKey = new Text("");

	private static final Log log = LogFactory.getLog(DliArticleEnrichmentReducer.class);

	@Override

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

		super.setup(context);

		System.out.println("LIMIT: " + LIMIT);

	}

	@Override

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

			throws IOException, InterruptedException {

		// TODO Auto-generated method stub

		super.reduce(key, values, context);

	}

	private void emit(final List<EventWrapper> eventWrappers, final Context context) {

		eventWrappers.stream().filter(Objects::nonNull).forEach(ew -> emit(ew, context));

	}

	private void emit(final EventWrapper eventWrapper, final Context context) {

		try {

			final Text valueout = new Text(eventWrapper.asBrokerEvent().asJson());

			context.write(tKey, valueout);

			eventWrapper.incrementCounter(context);

		} catch (Exception e) {

			throw new RuntimeException(e);

		}

	}

}