D-Net

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

import java.io.IOException;

import java.util.Map;

import eu.dnetlib.data.mapreduce.JobParams;

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

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

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

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

import eu.dnetlib.data.proto.OafProtos.OafRel.Builder;

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

import eu.dnetlib.data.transform.xml.AbstractDNetXsltFunctions;

import eu.dnetlib.pace.config.DedupConfig;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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;

public class DedupMarkDeletedEntityMapper extends TableMapper<ImmutableBytesWritable, Put> {

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

	private DedupConfig dedupConf;

	@Override

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

		dedupConf = DedupConfig.load(context.getConfiguration().get(JobParams.DEDUP_CONF));

		log.info("dedup findRoots mapper\nwf conf: " + dedupConf.toString());

	}

	@Override

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

		// System.out.println("Find root mapping: " + new String(rowkey.copyBytes()));

		final Type type = Type.valueOf(dedupConf.getWf().getEntityType());

		final Map<byte[], byte[]> mergedIn = value.getFamilyMap(DedupUtils.getDedupCF_mergedInBytes(type));

		if ((mergedIn != null) && !mergedIn.isEmpty()) {

			final byte[] row = rowkey.copyBytes();

			// marks the original body deleted

			emitBody(context, row, value.getValue(Bytes.toBytes(type.toString()), DedupUtils.BODY_B));

		} else {

			context.getCounter(type.toString(), "row not merged").increment(1);

		}

	}

	private void emitBody(final Context context, final byte[] row, final byte[] body) throws IOException, InterruptedException {

		final String type = dedupConf.getWf().getEntityType();

		if (body == null) {

			context.getCounter(type, "missing body").increment(1);

			System.err.println("missing body: " + new String(row));

			return;

		}

		final Oaf prototype = Oaf.parseFrom(body);

		if (prototype.getDataInfo().getDeletedbyinference()) {

			context.getCounter(type, "bodies already deleted").increment(1);

		} else {

			final Oaf.Builder oafRoot = Oaf.newBuilder(prototype);

			oafRoot.getDataInfoBuilder().setDeletedbyinference(true).setInferred(true).setInferenceprovenance(dedupConf.getWf().getConfigurationId());

			final byte[] family = Bytes.toBytes(type);

			final Put put = new Put(row).add(family, DedupUtils.BODY_B, oafRoot.build().toByteArray());

			put.setWriteToWAL(JobParams.WRITE_TO_WAL);

			context.write(new ImmutableBytesWritable(row), put);

			context.getCounter(type, "bodies marked deleted").increment(1);

		}

	}

	private byte[] buildRel(final byte[] from, final byte[] to, final Dedup.RelName relClass) {

		final Builder oafRel = DedupUtils.getDedup(dedupConf, new String(from), new String(to), relClass);

		final Oaf oaf =

				Oaf.newBuilder()

						.setKind(Kind.relation)

						.setLastupdatetimestamp(System.currentTimeMillis())

						.setDataInfo(

								AbstractDNetXsltFunctions.getDataInfo(null, "", "0.8", false, true).setInferenceprovenance(

										dedupConf.getWf().getConfigurationId())).setRel(oafRel)

				.build();

		return oaf.toByteArray();

	}

}