D-Net

import com.google.common.collect.Sets;

import eu.dnetlib.data.proto.RelTypeProtos;

import java.util.Set;

    public enum Type {

        valid,

        notvalid,

        fromsemrel,

        fromresult;

    }

    public static final int FROM_SEM_REL = 10;//for the relation to projects propagated thank to a semantic relation

    public static final int FROM_RESULT = 20;//for the relations to projects already owned by the result

    public final static String DATA_INFO_TYPE = "propagation";

    public final static String SCHEMA_NAME = "dnet:provenanceActions";

    public final static String SCHEMA_ID = "dnet:provenanceActions";

    public final static String DNET_COUNTRY_SCHEMA = "dnet:countries";

    public final static String DNET_RELATION_SCHEMA = "dnet:result_project_relations";

    public final static String CLASS_RELATION_ID = "propagation::project::semrel";

    public final static String CLASS_COUNTRY_ID = "propagation::country::instrepos";

    public final static int PROJECT = TypeProtos.Type.project.getNumber();//40

    public final static int DATASOURCE = TypeProtos.Type.datasource.getNumber();//10

    public final static int ORGANIZATION = TypeProtos.Type.organization.getNumber();//20

    public final static int PUBLICATION = TypeProtos.Type.result.getNumber();//50

    public final static RelTypeProtos.RelType REL_TYPE = RelTypeProtos.RelType.resultProject;

    public final static RelTypeProtos.SubRelType SUBREL_TYPE = RelTypeProtos.SubRelType.outcome;

    public static final String REL_PROJECT_RESULT = "produces";

    public static final String REL_RESULT_PROJECT = "isProducedBy";

    public static final String RELATION = REL_TYPE + "_" + SUBREL_TYPE + "_";

    public static final String OUTCOME_PRODUCEDBY = RELATION + REL_RESULT_PROJECT;

    public static final String[] DEFAULT_RELATION_SET = new String[]{"resultResult_supplement_isSupplementedBy"};

    public static final Set<String> DEFAULT_ALLOWED_DATASOURCES = Sets.newHashSet("pubsrepository::institutional");

    private Set<String> whiteList = Sets.newHashSet("10|opendoar____::300891a62162b960cf02ce3827bb363c");

    private Set<String> blackList = Sets.newHashSet("");

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

import com.google.protobuf.InvalidProtocolBufferException;

import eu.dnetlib.data.proto.*;

import org.apache.hadoop.hbase.client.Result;

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

import java.util.Map;

public final class Utils {

    public static  OafProtos.OafEntity getEntity(Result value, TypeProtos.Type type) throws InvalidProtocolBufferException {

        final Map<byte[],byte[]> map = value.getFamilyMap(Bytes.toBytes(type.toString()));

        final byte[] body = map.get(Bytes.toBytes("body"));

        if (body != null){

            OafProtos.Oaf oaf = OafProtos.Oaf.parseFrom(body);

            if(oaf.getDataInfo().getDeletedbyinference())

                return null;

            return oaf.getEntity();

        }

        return null;

    }

    public static FieldTypeProtos.DataInfo.Builder getDataInfo(String trust, String class_id,String schema_id,String schema_name, String data_info_type,String class_name) {

        FieldTypeProtos.DataInfo.Builder builder = FieldTypeProtos.DataInfo.newBuilder()

                .setInferred(true)

                .setProvenanceaction(getQualifier(class_id,schema_id,schema_name,class_name)                        )

                .setInferenceprovenance(data_info_type)

                .setTrust(trust);

        return builder;

    }

    public static FieldTypeProtos.Qualifier.Builder getQualifier(String class_id, String schema_id, String schema_name, String class_name){

        return FieldTypeProtos.Qualifier.newBuilder()

                .setClassid(class_id)

                .setClassname(class_name)

                .setSchemeid(schema_id)

                .setSchemename(schema_name);

    }

    public static FieldTypeProtos.Qualifier.Builder getCountry(String countryValue, String trust,String dnet_schema, String class_id,String schema_id,String schema_name, String data_info_type, String class_name) {

        final FieldTypeProtos.Qualifier.Builder country = FieldTypeProtos.Qualifier.newBuilder()

                .setClassid(countryValue)

                .setClassname(countryValue)

                .setSchemeid(dnet_schema)

                .setSchemename(dnet_schema);

        country.setDataInfo(getDataInfo(trust,class_id,schema_id,schema_name,data_info_type,class_name));//"Propagation of country information from datasources belonging to institutional repositories"));

        return country;

    }

}

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

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

import eu.dnetlib.data.proto.OafProtos;

import org.apache.commons.logging.LogFactory;

import java.util.List;

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

            keyOut.set(key);

        ResultIterator rh = null;

        try {

            rh = new ResultCountryIterator(values,key.getKeyType().get());

        } catch (NotValidResultSequenceException e) {

            context.getCounter(COUNTER_PROPAGATION,e.getMessage()).increment(1);

        context.getCounter(COUNTER_PROPAGATION,"institutional datasource").increment(1);

        while(rh.hasNext()){

            OafProtos.Oaf oafUpdate = rh.next().get(0);

            emit(context, oafUpdate.getEntity().getId(), JsonFormat.printToString(oafUpdate));

            context.getCounter(COUNTER_PROPAGATION, " added country to product ").increment(1);

        if (!rh.getPropagate()){

            context.getCounter(COUNTER_PROPAGATION, "resultId expected in ordering was not found" ).increment(1);

import java.util.List;

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

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

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

import eu.dnetlib.data.proto.OafProtos;

        ResultIterator rh = null;

        try {

            rh = new ResultCountryIterator(values,key.getKeyType().get());

        } catch (NotValidResultSequenceException e) {

            context.getCounter(COUNTER_PROPAGATION,e.getMessage()).increment(1);

        context.getCounter(COUNTER_PROPAGATION,"institutional datasource").increment(1);

        while(rh.hasNext()){

            try{

                List<OafProtos.Oaf> oap = rh.next();

                byte[] oafUpdate = oap.get(0).toByteArray();

                byte[] targetRowKey = Bytes.toBytes(oap.get(0).getEntity().getId());

                OafRowKeyDecoder.decode(targetRowKey);

                final Put put = new Put(targetRowKey).add(Bytes.toBytes("result"), Bytes.toBytes("update_" + System.nanoTime()), oafUpdate);

                keyOut.set(targetRowKey);

                context.write(keyOut, put);

                context.getCounter(COUNTER_PROPAGATION, "added country to product").increment(1);

            }catch(IllegalArgumentException e){

                context.getCounter(COUNTER_PROPAGATION,"not valid result id in result list for country propagation").increment(1);

        }

        if (!rh.getPropagate()){

            context.getCounter(COUNTER_PROPAGATION, "resultId expected in ordering was not found" ).increment(1);

package eu.dnetlib.data.mapreduce.hbase.propagation.country.institutionalrepositories;

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

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

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

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

import eu.dnetlib.data.proto.KindProtos;

import eu.dnetlib.data.proto.OafProtos;

import eu.dnetlib.data.proto.ResultProtos;

import eu.dnetlib.data.proto.TypeProtos;

import org.apache.hadoop.io.Text;

import java.util.ArrayList;

import java.util.Arrays;

import java.util.List;

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

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

public class ResultCountryIterator  extends ResultIterator {

    private String country;

    public ResultCountryIterator(final Iterable<Text> values, int key) throws NotValidResultSequenceException {

        super(values,key);

    }

    @Override

    protected void checkSequence() throws NotValidResultSequenceException {

        if (!it.hasNext()) {

            throw new NotValidResultSequenceException("Empty information for key");

        }

        final Value first = Value.fromJson(it.next().toString());

        trust = first.getTrust();

        if (!(first.getValue().equals(ZERO) || first.getValue().equals(ONE))) {

            if (key != TypeProtos.Type.datasource.getNumber()) {

                throw new NotValidResultSequenceException("First Element in reducer is not type of datasource,  but the organization exists");

            } else {

                while (it.hasNext()) {

                    resultId = Value.fromJson(it.next().toString()).getValue();

                    if (!resultId.startsWith("50|")) {

                        throw new NotValidResultSequenceException("ERROR ORDERING CHECK");

                    }

                }

                throw new NotValidResultSequenceException("WARNING: unexpected first element");

            }

        }

        //ensure we are dealing with an institutional repository

        if (first.getValue().equals(ONE)) {

            //context.getCounter(COUNTER_PROPAGATION, "institutional datasource").increment(1);

            if (!it.hasNext()) {

                throw new NotValidResultSequenceException("No information apart of type of datasource");

            }

            country = Value.fromJson(it.next().toString()).getValue(); // need to update the information for the country to each element in the iterator

            if (country.trim().length() != 2) {

                try {

                    Integer.parseInt(country.trim());

                } catch (Exception e) {

                }

                throw new NotValidResultSequenceException("Second element in reducer is not country");

            }

            boolean iterate = true;

            resultId = TERMINATOR;

            while(it.hasNext() && iterate){

                resultId = Value.fromJson(it.next().toString()).getValue();

                if (!resultId.startsWith(("50|"))) {

                    if (!resultId.equalsIgnoreCase(country)) {

                        propagate = false;

                        iterate = false;

                    }

                }else{

                    propagate = true;

                    iterate = false;

                }

            }

            if (!resultId.equals(TERMINATOR)){

                try {

                    OafRowKeyDecoder.decode(resultId);

                }catch(IllegalArgumentException e){

                    throw new NotValidResultSequenceException("No result in sequence");

                }

            }

        }else

            throw new NotValidResultSequenceException("Not allowed dsType institutional datasource");

    }

    @Override

    public List<OafProtos.Oaf> next() {

        final ResultProtos.Result.Metadata.Builder metadata = ResultProtos.Result.Metadata.newBuilder().addCountry(

                getCountry(country,trust,DNET_COUNTRY_SCHEMA,CLASS_COUNTRY_ID,SCHEMA_ID,SCHEMA_NAME,DATA_INFO_TYPE,"Propagation of country information from datasources belonging to institutional repositories"));

        final ResultProtos.Result.Builder result = ResultProtos.Result.newBuilder().setMetadata(metadata);

        final OafProtos.OafEntity.Builder entity = OafProtos.OafEntity.newBuilder()

                .setType(TypeProtos.Type.result)

                .setId(resultId)

                .setResult(result);

        if(it.hasNext())

            resultId=Value.fromJson(it.next().toString()).getValue();

        else

            resultId = TERMINATOR;

        return new ArrayList<OafProtos.Oaf>(Arrays.asList(OafProtos.Oaf.newBuilder()

                .setKind(KindProtos.Kind.entity)

                .setEntity(entity)

                .build()));

    }

}

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

                        context.getCounter(COUNTER_PROPAGATION, "emit hostedby | collectedfrom for publication ").increment(1);

                            context.getCounter(COUNTER_PROPAGATION, "emit hostedby | collectedfrom for publication ").increment(1);

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

public class NotValidResultSequenceException extends Exception {

    private String message;

    public NotValidResultSequenceException(String message){

        this.message = message;

    }

    @Override

    public String getMessage() {

        return message;

    }

}

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

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

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

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

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

import eu.dnetlib.data.proto.KindProtos;

import eu.dnetlib.data.proto.OafProtos;

import eu.dnetlib.data.proto.RelMetadataProtos;

import eu.dnetlib.data.proto.ResultProjectProtos;

import org.apache.commons.lang3.StringUtils;

import org.apache.hadoop.io.Text;

import java.util.*;

import java.util.stream.Collectors;

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

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

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

public class ResultProjectIterator extends ResultIterator {

    private Set<String> projectList;

    private Iterator<String> pli ;

    private final static String MSG = "Propagation of relation to project through semantic relation among results";

    public ResultProjectIterator(final Iterable<Text> values, final String key) throws NotValidResultSequenceException{

        super(values, key);

    }

    @Override

    protected void checkSequence() throws NotValidResultSequenceException {

        if(!it.hasNext()){

            throw new NotValidResultSequenceException("Empty information for key");

        }

        projectList = new HashSet<>();

        loadProjectList();

        pli = projectList.iterator();

        getNext();

    }

    private void getNext(){

        if (pli.hasNext())

            resultId = pli.next();

        else

            resultId = TERMINATOR;

    }

    private void loadProjectList() {

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

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

        while (it.hasNext()) {

            Value v = Value.fromJson(it.next().toString());

            if(trust == null) trust = v.getTrust();

            if (v.getType() == PropagationConstants.Type.fromresult) {

                if(StringUtils.isNotBlank(v.getValue()))

                    fromResult.addAll(Arrays.asList(StringUtils.split(v.getValue(), ",")));

            } else {

                fromSemRel.addAll(Arrays.asList(StringUtils.split(v.getValue(), ",")));

            }

        }

       //takes the projects to be associated to the result that have not already been associated to it

        projectList.addAll(fromSemRel.stream()

                .filter(it->!fromResult.contains(it))

                .collect(Collectors.toCollection((HashSet::new))));

    }

    @Override

    public List<OafProtos.Oaf> next() {

        ArrayList<OafProtos.Oaf> ret = new ArrayList<OafProtos.Oaf> (Arrays.asList(

                getOafRel(keyb,resultId,REL_RESULT_PROJECT),

                getOafRel(resultId,keyb,REL_PROJECT_RESULT)

        ));

        getNext();

        return ret;

    }

    private OafProtos.Oaf getOafRel(String source, String target, String semantics){

        final ResultProjectProtos.ResultProject.Builder rpb = ResultProjectProtos.ResultProject.newBuilder()

                .setOutcome(

                        ResultProjectProtos.ResultProject.Outcome.newBuilder()

                                .setRelMetadata(

                                        RelMetadataProtos.RelMetadata.newBuilder()

                                                .setSemantics(

                                                        getQualifier(semantics,semantics,DNET_RELATION_SCHEMA,DNET_RELATION_SCHEMA)

                                                )

                                )

                );

        final OafProtos.OafRel.Builder relation = OafProtos.OafRel.newBuilder()

                .setChild(false)

                .setSubRelType(SUBREL_TYPE)

                .setRelType(REL_TYPE)

                .setRelClass(semantics)

                .setTarget(target)

                .setSource(source)

                .setResultProject(rpb);

        return OafProtos.Oaf.newBuilder().setKind(KindProtos.Kind.relation)

                .setRel(relation)

                .setDataInfo(getDataInfo(trust,CLASS_RELATION_ID,SCHEMA_ID,SCHEMA_NAME,DATA_INFO_TYPE,MSG))

                .build();

    }

}

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

import com.google.protobuf.InvalidProtocolBufferException;

import eu.dnetlib.data.proto.OafProtos.Oaf;

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

import java.io.IOException;

import java.util.Map;

public class SemanticRelationCounter extends TableMapper<ImmutableBytesWritable, Writable> {

    private String[] sem_rels;

    @Override

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

        super.setup(context);

        String[] default_set = {"resultResult_supplement_isSupplementedBy"};

        sem_rels = context.getConfiguration().getStrings("propagatetoproject.semanticrelations",default_set);

        if (sem_rels.length == 0)

            throw new InterruptedException("No semantic relation over which propagate is specified");

    }

    @Override

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

        final Map<byte[], byte[]> resultMap = value.getFamilyMap(Bytes.toBytes("result"));

        final byte[] body = resultMap.get(Bytes.toBytes("body"));

        if (body != null) {

            final Oaf res_oaf = Oaf.parseFrom(body);

            if (res_oaf.getDataInfo().getDeletedbyinference()){

                context.getCounter("Relation counter","del by inference result").increment(1);

                return;

            }

            context.getCounter("Relation counter", "not null body ").increment(1);

            boolean foundRelation = false;

            for (String sem_rel : sem_rels)

                foundRelation = foundRelation || countRelation(sem_rel,context,value);

            if(countRelation("resultProject_outcome_isProducedBy",context,value)){

                if (foundRelation)

                    context.getCounter("Relation counter", "Association to project in result with semantic relation" ).increment(1);

            }

        }

    }

    private boolean countRelation(String rel_to_find, Context context, Result value) throws InvalidProtocolBufferException {

        final Map<byte[], byte[]> relationMap = value.getFamilyMap(Bytes.toBytes(rel_to_find));

        boolean ret = false;

        for(byte[] relation:relationMap.values() ){

            final Oaf rel_oaf = Oaf.parseFrom(relation);

            if (!isValid(rel_oaf)) {

                context.getCounter("Relation counter", rel_to_find + " not present in relation set").increment(1);

                return ret;

            }

            if (rel_oaf.getDataInfo().getDeletedbyinference()){

                context.getCounter("Relation counter ",rel_to_find + " del by inference" ).increment(1);

            }else{

                context.getCounter("Relation counter ",rel_to_find ).increment(1);

                ret = true;

            }

        }

        return ret;

    }

    private boolean isValid(final Oaf oaf) {

        return (oaf != null) && oaf.isInitialized();

    }

}

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

import com.google.common.collect.ComparisonChain;

import eu.dnetlib.data.mapreduce.hbase.propagation.projecttoresult.ProjectToResultKey;

import org.apache.hadoop.io.IntWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.io.WritableComparable;

import java.io.DataInput;

import java.io.DataOutput;

import java.io.IOException;

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

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

public class ProjectToResultKey implements WritableComparable<ProjectToResultKey> {

    private IntWritable keyType;

    private Text id;

    public ProjectToResultKey() {

    }

    public static ProjectToResultKey create(final int keyType, final String id) {

        return new ProjectToResultKey(keyType, id);

    }

    public static ProjectToResultKey existSemRel(final String id) {

        return new ProjectToResultKey(FROM_SEM_REL, id);

    }

    public static ProjectToResultKey doesNotExistSemRel(final String id) {

        return new ProjectToResultKey(FROM_RESULT, id);

    }

    public ProjectToResultKey(final int keyType, final String id) {

        this.id = new Text(id);

        this.keyType = new IntWritable(keyType);

    }

    public void setKeyType(final IntWritable keyType) {

        this.keyType = keyType;

    }

    public void setId(final Text id) {

        this.id = id;

    }

    public Text getId() {

        return id;

    }

    public IntWritable getKeyType() {

        return keyType;

    }

    @Override

    public int compareTo(final ProjectToResultKey o) {

        final int res = ComparisonChain.start()

                .compare(getId(), o.getId())

                .compare(getKeyType(), o.getKeyType())

                .result();

        //System.out.println(String.format("%s.compareTo(%s) = %s", toString(), o.toString(), res));

        return res;

    }

    @Override

    public void write(final DataOutput out) throws IOException {

        keyType.write(out);

        id.write(out);

    }

    @Override

    public void readFields(final DataInput in) throws IOException {

        keyType = new IntWritable();

        keyType.readFields(in);

        id = new Text();

        id.readFields(in);

    }

    @Override

    public String toString() {

        return (new StringBuilder())

                .append('{')

                .append(getKeyType().get())

                .append(',')

                .append(getId())

                .append('}')

                .toString();

    }

}

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

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

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

import eu.dnetlib.data.proto.*;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.apache.hadoop.hbase.client.Put;

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

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

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

import org.apache.hadoop.io.Text;

import java.io.IOException;

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

public class PropagationProjectToResultReducer extends TableReducer<ImmutableBytesWritable, Text, ImmutableBytesWritable> {

    private static final Log log = LogFactory.getLog(PropagationProjectToResultReducer.class); // NOPMD by marko on 11/24/08 5:02 PM

    private ImmutableBytesWritable keyOut;

    @Override

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

        super.setup(context);

        keyOut = new ImmutableBytesWritable();

    }

    @Override

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

        ResultIterator rh = null;

        try {

            rh = new ResultProjectIterator(values,Bytes.toString(key.copyBytes()));

        } catch (NotValidResultSequenceException e) {

            context.getCounter(COUNTER_PROPAGATION,e.getMessage()).increment(1);

            return;

        }

        while(rh.hasNext()){

            for(OafProtos.Oaf oaf:rh.next()){

                final String source = oaf.getRel().getSource();

                final Put put = new Put(Bytes.toBytes(source)).add(Bytes.toBytes(RELATION + oaf.getRel().getRelClass()),Bytes.toBytes(oaf.getRel().getTarget()),oaf.toByteArray());

                keyOut.set(Bytes.toBytes(source));

                context.write(keyOut, put);

            }

            context.getCounter(COUNTER_PROPAGATION,"Added relation to project").increment(1);

        }

    }

}

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

import com.googlecode.protobuf.format.JsonFormat;

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

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

import eu.dnetlib.data.proto.*;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

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

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

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

public class PropagationProjectToResultFileReducer extends Reducer<ImmutableBytesWritable, Text, Text,Text> {

    private static final Log log = LogFactory.getLog(PropagationProjectToResultReducer.class); // NOPMD by marko on 11/24/08 5:02 PM

    private Text keyOut;

    private Text outValue;

    @Override

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

        super.setup(context);

        keyOut = new Text("");

        outValue = new Text();

    }

    @Override

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

        ResultIterator rh = null;

        try {

            rh = new ResultProjectIterator(values, Bytes.toString(key.copyBytes()));

        } catch (NotValidResultSequenceException e) {

            context.getCounter(COUNTER_PROPAGATION, e.getMessage()).increment(1);

            return;

        }

        while (rh.hasNext()) {

            for (OafProtos.Oaf oaf : rh.next()) {

                keyOut.set(oaf.getRel().getTarget());

                outValue.set(JsonFormat.printToString(oaf).getBytes());

                context.write(keyOut, outValue);

            }

            context.getCounter(COUNTER_PROPAGATION, "Added relation to project").increment(1);

        }

    }

}

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

import eu.dnetlib.data.mapreduce.hbase.propagation.country.institutionalrepositories.InstOrgKey;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Partitioner;

/**

 * Created by miriam on 17/08/2018.

 */

public class NaturalProjectToResultKeyPartitioner extends Partitioner<ProjectToResultKey, Text > {

    @Override

    public int getPartition(ProjectToResultKey key, Text val, int numPartitions) {

        final int res = Math.abs(key.getId().hashCode() % numPartitions);

        return res;

    }

}

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

import eu.dnetlib.data.mapreduce.hbase.propagation.country.institutionalrepositories.InstOrgKey;

import org.apache.hadoop.io.WritableComparable;

import org.apache.hadoop.io.WritableComparator;

/**

 * Created by miriam on 17/08/2018.

 */

public class NaturalProjectToResultKeyGroupingComparator extends WritableComparator {

    protected NaturalProjectToResultKeyGroupingComparator() {

        super(ProjectToResultKey.class, true);

    }

    @Override

    public int compare(WritableComparable w1, WritableComparable w2) {

        final InstOrgKey k1 = (InstOrgKey) w1;

        final InstOrgKey k2 = (InstOrgKey) w2;

        return k1.getId().compareTo(k2.getId());

    }

}

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

import com.google.protobuf.InvalidProtocolBufferException;

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

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

import eu.dnetlib.data.proto.OafProtos;

import eu.dnetlib.data.proto.TypeProtos;

import org.apache.commons.lang3.StringUtils;

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 java.io.IOException;

import java.util.*;

import java.util.stream.Collectors;

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

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

//public class ProjectToResultMapper extends TableMapper<ProjectToResultKey, Text> {

public class ProjectToResultMapper extends TableMapper<ImmutableBytesWritable, Text> {

    private static final String SEPARATOR = ",";

    private String[] sem_rels;

    private String trust;

    private ImmutableBytesWritable keyOut;

    private Text valueOut;

    @Override

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

        keyOut = new ImmutableBytesWritable();

        valueOut = new Text();

        sem_rels = context.getConfiguration().getStrings("propagatetoproject.semanticrelations", DEFAULT_RELATION_SET);

        trust = context.getConfiguration().get("propagatetoproject.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;

        }

        context.getCounter(COUNTER_PROPAGATION, "Valid result ").increment(1);

        //selection of all the projects associated to this result

        String projectIdList = getProjectIdList(value);

        //if the list of projects is not empty, verify if it exists some allowed semantic relation to which propagate the project

        if(StringUtils.isNotBlank(projectIdList)){

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

            //selection of all the results bind by this result considering all the allowed semantic relations

            for (String sem_rel : sem_rels)

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

            emit(context, toemitrelations, projectIdList);

            context.getCounter(COUNTER_PROPAGATION, "emit for semantic relation").increment(toemitrelations.size());

        }

        keyOut.set(entity.getId().getBytes());

        valueOut.set(Value.newInstance(projectIdList,Type.fromresult).toJson());

        //context.write(ProjectToResultKey.create(FROM_RESULT,entity.getId()), valueOut);

        context.write(keyOut, valueOut);

        context.getCounter(COUNTER_PROPAGATION, "emit for result").increment(1);

    }

    //emit for each valid semantic relation the id of the relation target and the list of projects associated to the source of the relation

    private void emit( Context context, Set<String> toemitrelations, String projectIdList) throws IOException, InterruptedException {

        for(String relation : toemitrelations){

            keyOut.set(relation.getBytes());

            valueOut.set(Value.newInstance( projectIdList,trust,Type.fromsemrel).toJson());

            context.write(keyOut, valueOut);

        }

    }

    //starting from the Result gets the list of projects it is related to and returns it as a csv

    private String getProjectIdList(Result value) throws InvalidProtocolBufferException {

        Set<String> ret = getRelationTarget(value, OUTCOME_PRODUCEDBY);

        return ret.size() == 0 ? null : String.join(SEPARATOR, ret);

    }

    private Set<String> getRelationTarget(Result value, String sem_rel) throws InvalidProtocolBufferException {

        final Map<byte[], byte[]> relationMap = value.getFamilyMap(Bytes.toBytes(sem_rel));

        /*

        we could extract the target qualifiers from the familyMap's keyset, but we also need to check the relationship is not deletedbyinference

        return relationMap.keySet().stream()

                .map(String::new)

                .collect(Collectors.toCollection(HashSet::new));

        */

        return relationMap.values().stream()

                .map(this::asOaf)

                .filter(Objects::nonNull)

                .filter(o -> isValid(o))

                .filter(o -> !o.getDataInfo().getDeletedbyinference())

                .map(o -> o.getRel().getTarget())

                .collect(Collectors.toCollection(HashSet::new));

    }

    private OafProtos.Oaf asOaf(byte[] r) {

        try {

            return OafProtos.Oaf.parseFrom(r);

        } catch (InvalidProtocolBufferException e) {

            return null;

        }

    }

    private boolean isValid(final OafProtos.Oaf oaf) {

        return (oaf != null) && oaf.isInitialized();

    }

}

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

import eu.dnetlib.data.proto.OafProtos;

import org.apache.hadoop.io.Text;

import java.util.Iterator;

import java.util.List;

public abstract class ResultIterator implements Iterator<List<OafProtos.Oaf>>{

    protected Iterator<Text> it;

    protected int key;

    protected String keyb;

    protected String resultId;

    protected boolean propagate = true;

    protected final static String TERMINATOR = "FINITO";

    protected String trust = null;

    public ResultIterator(final Iterable<Text> values, int key) throws NotValidResultSequenceException {

        it = values.iterator();

        this.key = key;

        checkSequence();

    }

    public ResultIterator(final Iterable<Text> values, final String key) throws NotValidResultSequenceException {

        it = values.iterator();

        this.keyb = key;

        checkSequence();

    }

    protected abstract void checkSequence() throws NotValidResultSequenceException ;

    public abstract List<OafProtos.Oaf> next() ;

    public boolean hasNext(){

        return (!resultId.equals(TERMINATOR) && propagate);

    }

    public boolean getPropagate(){

        return propagate;

    }

    public String getResultId(){

        return resultId;

    }

}

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

    private Type type;

    public Value(String value, Type type) {

        this.value = value;

        this.type=type;

    }

    public Value(String value, String trust, Type type) {

        this.value=value;

        this.trust=trust;

        this.type=type;

    }

    public static Value newInstance(String value,Type type) {

        return new Value(value,type);

    }

    public static Value newInstance(String value, String trust, Type type){

        return new Value(value,trust,type);

    }

    public Value setValue(String value) {

        return this;

    public Value setTrust(String trust) {

        return this;

    public Type getType() {

        return type;

    }

    public Value setType(Type type) {

        this.type = type;

        return this;

    }