D-Net

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

import java.io.IOException;

import java.util.List;

import eu.dnetlib.data.mapreduce.JobParams;

import eu.dnetlib.data.mapreduce.util.dao.HBaseTableDAO;

import eu.dnetlib.data.proto.DNGFProtos.DNGF;

import eu.dnetlib.data.proto.FieldTypeProtos;

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

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 static eu.dnetlib.data.mapreduce.util.dao.HBaseTableDAO.getMetadata;

public class DedupMarkDeletedEntityMapper extends TableMapper<ImmutableBytesWritable, Put> {

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

    private static final String DEFAULT_PROVENANCE = "dnet";

	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 List<String> mergedIn = HBaseTableDAO.getTargetIds(value, "isMergedIn");

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

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

			// marks the original body deleted

			emitBody(context, row, getMetadata(value, type));

		} else {

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

		}

	}

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

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

		if (prototype == null) {

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

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

			return;

		}

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

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

		} else {

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

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

            if (!oafRoot.getDataInfoBuilder().hasProvenanceaction()) {

                context.getCounter(type, "missing provenance, setting " + DEFAULT_PROVENANCE).increment(1);

                oafRoot.getDataInfoBuilder()

                        .setProvenanceaction(FieldTypeProtos.Qualifier.newBuilder()

                                .setClassname(DEFAULT_PROVENANCE).setClassid(DEFAULT_PROVENANCE).setSchemename(DEFAULT_PROVENANCE).setSchemeid(DEFAULT_PROVENANCE));

            }

			context.write(new ImmutableBytesWritable(row), HBaseTableDAO.asPut(oafRoot.build()));

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

		}

	}

}