D-Net

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

import java.io.IOException;

import java.nio.charset.Charset;

import java.util.Set;

import org.apache.commons.lang3.StringUtils;

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 com.google.common.collect.Iterables;

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

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

public abstract class AbstractOpenOrgsMapper extends TableMapper<ImmutableBytesWritable, ImmutableBytesWritable> {

	private ImmutableBytesWritable keyOut;

	private ImmutableBytesWritable valueOut;

	@Override

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

		super.setup(context);

		keyOut = new ImmutableBytesWritable();

		valueOut = new ImmutableBytesWritable();

	}

	@Override

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

		try {

			if (DedupUtils.isRoot(keyIn)) {

				context.getCounter("organization", "dedupwf_____ row skipped").increment(1);

				return;

			}

			if (OpenOrgsCommon.isOpenOrgsMesh(keyIn)) {

				context.getCounter("organization", "openorgsmesh row skipped").increment(1);

				return;

			}

			final byte[] bytes = value.getFamilyMap(Bytes.toBytes("organization")).get(Bytes.toBytes("body"));

			if (bytes == null) {

				context.getCounter("organization", "empty body").increment(1);

				return;

			}

			final Set<byte[]> roots = value.getFamilyMap(Bytes.toBytes("organizationOrganization_dedup_isMergedIn")).keySet();

			if (roots.isEmpty()) {

				context.getCounter("organization", "missing mergedIn relationship").increment(1);

			}

			if (roots.size() > 1) {

				context.getCounter("organization", "too many mergedIn relationship").increment(1);

			}

			final String root = roots.isEmpty() ? null : new String(Iterables.getOnlyElement(roots), Charset.forName("UTF-8"));

			final String keyOutString = calculateKeyOut(Oaf.parseFrom(bytes), root, context);

			if (StringUtils.isNotBlank(keyOutString)) {

				emit(keyOutString, bytes, context);

			}

		} catch (final Exception e) {

			throw new RuntimeException(e);

		}

	}

	protected abstract String calculateKeyOut(Oaf oaf, String rootId, Context context);

	private void emit(final String key, final byte[] value, final Context context) {

		keyOut.set(key.getBytes());

		valueOut.set(value);

		try {

			context.write(keyOut, valueOut);

		} catch (IOException | InterruptedException e) {

			throw new RuntimeException(e);

		}

	}

}