D-Net

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

import java.io.IOException;

import java.util.Map.Entry;

import java.util.stream.Collectors;

import eu.dnetlib.data.graph.model.DNGFRelDecoder;

import eu.dnetlib.data.graph.utils.RelDescriptor;

import eu.dnetlib.data.mapreduce.JobParams;

import eu.dnetlib.data.proto.DNGFProtos;

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

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

import eu.dnetlib.data.proto.FieldTypeProtos;

import eu.dnetlib.data.proto.TypeProtos;

import eu.dnetlib.data.transform.Ontologies;

import eu.dnetlib.data.transform.OntologyLoader;

import eu.dnetlib.pace.config.DedupConfig;

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.io.Writable;

import org.springframework.beans.factory.annotation.Qualifier;

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

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

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

public class DedupMergeRelationVersionMapper extends TableMapper<ImmutableBytesWritable, Writable> {

	private DedupConfig dedupConf;

	private ImmutableBytesWritable outKey = new ImmutableBytesWritable();

    private Ontologies ontologies;

	@Override

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

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

		System.out.println("dedup findRoots mapper\nwf conf: " + dedupConf.toString());

		outKey = new ImmutableBytesWritable();

        ontologies = OntologyLoader.loadOntologies(context.getConfiguration().get(JobParams.ONTOLOGIES));

        System.out.println("ontologies: " + ontologies.toJson(true));

    }

	@Override

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

        if (!isRoot(rowkey)) {

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

            return;

        }

        relVersions(

				value,

				"isMergedIn", "merges", "isSimilarTo")

				.entrySet().stream()

				.filter(e -> e.getValue().keySet().size() > 1)

				.collect(Collectors.toMap(

						e -> e.getKey(),

						e -> {

                            final DNGF.Builder b = DNGF.newBuilder();

                            e.getValue().values().forEach(v -> {

								b.mergeFrom(v);

								final String relType = RelDescriptor.asString(v.getRel().getRelType());

								context.getCounter("merged", relType).increment(1);

							});

                            return b;

                        }

				)).entrySet().forEach(dngf -> emit(context, rowkey, dngf));

	}

    private void emit(final Context context, final ImmutableBytesWritable rowkey, final Entry<String, DNGF.Builder> entry) {

        try {

			outKey.set(rowkey.copyBytes());

            DNGF.Builder inputRel = entry.getValue();

            System.out.println(String.format("wirting row: %s, data: %s", new String(rowkey.copyBytes()), inputRel.toString()));

            context.write(outKey, asPut(inputRel.build()));

            final String target = inputRel.getRel().getTarget();

            final TypeProtos.Type targetType = inputRel.getRel().getTargetType();

            final TypeProtos.Type sourceType = inputRel.getRel().getSourceType();

            final DNGFProtos.DNGFRel.Builder relBuilder = DNGFProtos.DNGFRel.newBuilder(inputRel.getRel());

            //SWAP Source Target and its type

            relBuilder.setTargetType(sourceType);

            relBuilder.setSourceType(targetType);

            relBuilder.setTarget(relBuilder.getSource());

            relBuilder.setSource(target);

            //Change relation to its inverse

            final String inverseRelation = ontologies.inverseOf(DNGFRelDecoder.decode(inputRel.getRel()).getRelDescriptor());

            relBuilder.setRelType(FieldTypeProtos.Qualifier.newBuilder(relBuilder.getRelType()).setClassname(inverseRelation).setClassid(inverseRelation).build());

            outKey.set(relBuilder.getSource().getBytes());

            inputRel.setRel(relBuilder.build());

            context.write(outKey, asPut(inputRel.build()));

		} catch (Exception e) {

			throw new RuntimeException(e);

		}

	}

}