D-Net

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

import java.io.IOException;

import java.util.Iterator;

import com.google.common.collect.Iterables;

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

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

import eu.dnetlib.data.mapreduce.JobParams;

import eu.dnetlib.data.mapreduce.hbase.dedup.kv.DNGFKey;

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

import eu.dnetlib.data.proto.DNGFProtos;

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

import eu.dnetlib.data.proto.FieldTypeProtos;

import eu.dnetlib.data.transform.Ontologies;

import eu.dnetlib.data.transform.OntologyLoader;

import eu.dnetlib.pace.config.DedupConfig;

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

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

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

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

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

/**

 * Created by sandro on 2/24/17.

 */

public class DedupFixRelationReducer extends TableReducer<DNGFKey, ImmutableBytesWritable, ImmutableBytesWritable> {

    public static final String COUNTER_GROUP = "Fix relations";

    private DedupConfig dedupConf;

    private Ontologies ontologies;

    @Override

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

        super.setup(context);

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

        System.out.println("dedup fix Relations reducer\n\nwf conf: " + dedupConf.toString());

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

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

    }

    @Override

    protected void reduce(DNGFKey key, Iterable<ImmutableBytesWritable> values, Context context) throws IOException, InterruptedException {

        if (isRoot(key.toString())) {

            System.err.println("aborting DedupFixRelationReducer, found root key: " + key);

            context.getCounter(COUNTER_GROUP, "aborted").increment(1);

            return;

        }

        final Iterator<ImmutableBytesWritable> it = values.iterator();

        final DNGF first = DNGFDecoder.decode(it.next().copyBytes()).getDNGF();

        if (!first.getRel().getRelType().getClassid().equals("merges")) {

            context.getCounter(COUNTER_GROUP, "Relation skipped").increment(1);

            return;

        }

        context.getCounter(COUNTER_GROUP, "Item to fix").increment(1);

        final String dedupRoot = first.getRel().getSource();

        it.forEachRemaining(b -> {

            try {

                handleRels(context, DNGFDecoder.decode(b.copyBytes()).getDNGF(), dedupRoot);

            } catch (Exception e) {

               throw new RuntimeException(e);

            }

        });

    }

    private void handleRels(final Context context, final DNGFProtos.DNGF dngf, final String dedupRoot) throws IOException, InterruptedException {

        final String relType = dngf.getRel().getRelType().getClassid();

        if (relType.contains("merges")) {

            return;

        }

        //Set relation deleted by inference from Root to entity that has been merged to another one

        final FieldTypeProtos.DataInfo.Builder dataInfoBuilder = FieldTypeProtos.DataInfo.newBuilder(dngf.getDataInfo());

        dataInfoBuilder.setDeletedbyinference(true);

        DNGFProtos.DNGF.Builder dngfBuilder = DNGFProtos.DNGF.newBuilder(dngf);

        dngfBuilder.setDataInfo(dataInfoBuilder.build());

        final String sourceKey = dngf.getRel().getSource();

        context.write(new ImmutableBytesWritable(Bytes.toBytes(sourceKey)), asPut(dngfBuilder.build()));

        context.getCounter(COUNTER_GROUP, "Relation set deleted").increment(1);

        // Create Relation from Root Entity to its deduplicated Entity

        dngfBuilder = DNGFProtos.DNGF.newBuilder(dngf);

        DNGFProtos.DNGFRel.Builder relBuilder = DNGFProtos.DNGFRel.newBuilder(dngf.getRel());

        relBuilder.setTarget(dedupRoot);

        dngfBuilder.setRel(relBuilder.build());

        context.write(new ImmutableBytesWritable(Bytes.toBytes(sourceKey)), asPut(dngfBuilder.build()));

        //Create Relation from deduplicated Entity to Root Entity

        relBuilder = DNGFProtos.DNGFRel.newBuilder(dngf.getRel());

        relBuilder.setTarget(relBuilder.getSource());

        relBuilder.setSource(dedupRoot);

        final String inverseRelation = HBaseTableDAO.getInverseRelation(DNGFRelDecoder.decode(dngf.getRel()).getRelDescriptor(), ontologies);

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

        dngfBuilder.setRel(relBuilder.build());

        context.write(new ImmutableBytesWritable(Bytes.toBytes(dedupRoot)), asPut(dngfBuilder.build()));

        context.getCounter(COUNTER_GROUP, "Relation fixed").increment(2);

    }

    private Iterable<DNGFProtos.DNGF> toDNGF(final Iterable<ImmutableBytesWritable> values) {

        return Iterables.transform(values, ibw -> DNGFDecoder.decode(ibw.copyBytes()).getDNGF());

    }

}