D-Net

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

import eu.dnetlib.data.bulktag.CommunityConfiguration;

import eu.dnetlib.data.bulktag.CommunityConfigurationFactory;

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

import org.apache.commons.lang.StringUtils;

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

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

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

import org.apache.hadoop.hbase.mapreduce.TableMapper;

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

import org.apache.hadoop.io.Writable;

import java.io.IOException;

import java.util.Map;

public class BulkTaggingMapper extends TableMapper<ImmutableBytesWritable, Writable> {

	private CommunityConfiguration cc;

	private ResultTagger tagger;

	@Override

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

		super.setup(context);

		final String conf = context.getConfiguration().get("tagging.conf");

		if (StringUtils.isBlank(conf)) {

			throw new IllegalArgumentException("missing bulk tagging configuration");

		}

		cc = CommunityConfigurationFactory.fromJson(conf);

		tagger = new ResultTagger();

	}

	@Override

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

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

		final byte[] body = resultMap.get(Bytes.toBytes("body"));

		if (body != null) {

			final Oaf oaf = tagger.enrichContext(Oaf.parseFrom(body), cc, context);

			if (oaf == null)

				return;

			final Put put = new Put(key.copyBytes()).add(Bytes.toBytes("result"), Bytes.toBytes("body"), oaf.toByteArray());

			context.write(key, put);

		}

	}

}