D-Net

package eu.dnetlib.data.mapreduce.hbase.propagation.communitytoresult;

import com.google.common.collect.Maps;

import com.google.common.reflect.TypeToken;

import com.google.gson.Gson;

import com.google.protobuf.InvalidProtocolBufferException;

import eu.dnetlib.data.mapreduce.hbase.propagation.PropagationConstants;

import eu.dnetlib.data.proto.FieldTypeProtos;

import eu.dnetlib.data.proto.OafProtos;

import eu.dnetlib.data.proto.ResultProtos;

import eu.dnetlib.data.proto.TypeProtos;

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

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

import static eu.dnetlib.data.mapreduce.hbase.propagation.Utils.getRelationTarget;

import java.io.IOException;

import java.util.*;

import java.util.stream.Collectors;

public class CommunityToResultMapper extends TableMapper<ImmutableBytesWritable, Text> {

    private ImmutableBytesWritable keyOut;

    private Text valueOut;

    private String[] sem_rels;

    private String trust;

    @Override

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

        valueOut = new Text();

        sem_rels = context.getConfiguration().getStrings("propagatetocommunity.semanticrelations", DEFAULT_COMMUNITY_RELATION_SET);

        trust = context.getConfiguration().get("propagatetocommunity.trust","0.85");

    }

    @Override

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

        final TypeProtos.Type type = OafRowKeyDecoder.decode(keyIn.copyBytes()).getType();

        //If the type is not result I do not need to process it

        }

        final OafProtos.OafEntity entity = getEntity(value, type);

        if (entity == null) {

        }

        //verify if we have some relation

        for (String sem_rel : sem_rels)

            toemitrelations.addAll(getRelationTarget(value, sem_rel, context, COUNTER_PROPAGATION));

            context.getCounter(COUNTER_PROPAGATION, "No allowed semantic relation present in result").increment(1);

            return;

        Set<String> contextIds = entity.getResult().getMetadata().getContextList()

                .stream()

                .map(ResultProtos.Result.Context::getId)

                .collect(Collectors.toSet());

        NavigableMap<byte[], byte[]> map = value.getFamilyMap(Bytes.toBytes(TypeProtos.Type.result.toString()));

        for (Map.Entry<byte[], byte[]> e : map.entrySet()) {

            stringMap.put(Bytes.toString(e.getKey()), e.getValue());

        }

        for (final String o : stringMap.keySet()) {

            if (o.startsWith("update_")) {

                final OafProtos.Oaf update = OafProtos.Oaf.parseFrom(map.get(o));

                contextIds.addAll(update.getEntity().getResult().getMetadata().getContextList()

                        .stream()

                        .map(ResultProtos.Result.Context::getId)

                        .map(s -> s.split("::")[0])

                        .collect(Collectors.toSet()));

            }

        }

        if (contextIds.isEmpty()) {

            context.getCounter(COUNTER_PROPAGATION, "No context in the body and in the update of the result").increment(1);

            return;

        for (String id : idCommunityList) {

            if (contextIds.contains(id)) {

                for (String target : toemitrelations) {

                    keyOut.set(target.getBytes());

                    valueOut.set(Value.newInstance(id).setTrust(trust).toJson());

                    context.write(keyOut, valueOut);

                    context.getCounter(COUNTER_PROPAGATION, "Emit propagation for " + id).increment(1);