D-Net

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

import com.google.common.collect.Lists;

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.mapreduce.hbase.propagation.Value;

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

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

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.client.Result;

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

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

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

import org.apache.hadoop.io.Text;

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

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

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

import java.io.IOException;

import java.lang.reflect.Type;

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;

    List<String> idCommunityList;

    @Override

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

        Type listType = new TypeToken<List<String>>() {}.getType();

        idCommunityList = new Gson().fromJson(context.getConfiguration().get("community.id.list"),listType);

        keyOut = new ImmutableBytesWritable();

        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

        if (!type.equals(TypeProtos.Type.result)) {

            return;

        }

        //verify if entity is valid

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

        if (entity == null) {

            context.getCounter(COUNTER_PROPAGATION, "Del by inference or null body for result").increment(1);

            return;

        }

        final Set<String> toemitrelations = new HashSet<>();

        //verify if we have some relation

        for (String sem_rel : sem_rels)

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

        if (toemitrelations.isEmpty()) {

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

            return;

        }

        //verify if we have a relation to a context in the body

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

                .stream()

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

                .collect(Collectors.toSet());

        //verify if we have a relation to a context in the update part made by the inference

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

        final Map<String, byte[]> stringMap = Maps.newHashMap();

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

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

        }

        // we fetch all the body updates

        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()));

            }

        }

        //we verify if we have something

        if (contextIds.isEmpty()) {

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

            return;

        }

        //verify if some of the context collected for the result are associated to a community in the communityIdList

        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);

                }

            }

        }

    }

}