D-Net

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

import java.io.IOException;

import java.util.List;

import java.util.Map;

import java.util.stream.Collectors;

import com.google.common.collect.Iterables;

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

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

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

import eu.dnetlib.data.proto.DedupProtos.Dedup.RelName;

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

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

import org.apache.commons.collections.MapUtils;

import org.apache.hadoop.hbase.client.Result;

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

import org.apache.hadoop.hbase.util.Bytes;

/**

 * Created by claudio on 08/07/16.

 */

public class EnrichmentMapper extends AbstractEnrichmentMapper {

	@Override

	protected String counterGroup() {

		return "Broker Enrichment";

	}

	@Override

	protected void map(final ImmutableBytesWritable key, final Result value, final Context context) throws IOException, InterruptedException {

		final String type = OafRowKeyDecoder.decode(key.copyBytes()).getType().toString();

		final Map<byte[], byte[]> result = value.getFamilyMap(Bytes.toBytes(Type.result.name()));

		if (MapUtils.isEmpty(result)) {

			context.getCounter(counterGroup(), type + ": empty family map").increment(1);

			return;

		}

		final Oaf body = UpdateMerger.mergeBodyUpdates(context, result);

		if (body == null) {

			context.getCounter(counterGroup(), type + ": body null").increment(1);

			return;

		}

		final String mergedInCF = DedupUtils.getDedupCF_mergedIn(Type.result);

		final Map<byte[], byte[]> mergedIn = value.getFamilyMap(Bytes.toBytes(mergedInCF));

		final byte[] outKey = getEmitKey(context, key, mergedIn);

		emit(context, outKey, body.toByteArray(), type);

	}

}