classes test classes and resources for production of the actionset to include bipFinder score in results

dhp-workflows/dhp-aggregation/src/main/java/eu/dnetlib/dhp/actionmanager/bipfinder/BipDeserialize.java

@@ -0,0 +1,23 @@
+package eu.dnetlib.dhp.actionmanager.bipfinder;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+
+
+public class BipDeserialize extends HashMap<String, List<Score>> implements Serializable  {
+
+    public BipDeserialize(){
+        super();
+    }
+
+    public List<Score> get(String key) {
+
+        if (super.get(key) == null) {
+            return new ArrayList<>();
+        }
+        return super.get(key);
+    }
+
+}

dhp-workflows/dhp-aggregation/src/main/java/eu/dnetlib/dhp/actionmanager/bipfinder/BipScore.java

@@ -0,0 +1,25 @@
+package eu.dnetlib.dhp.actionmanager.bipfinder;
+
+import java.io.Serializable;
+import java.util.List;
+
+public class BipScore implements Serializable {
+    private String id;
+    private List<Score> scoreList;
+
+    public String getId() {
+        return id;
+    }
+
+    public void setId(String id) {
+        this.id = id;
+    }
+
+    public List<Score> getScoreList() {
+        return scoreList;
+    }
+
+    public void setScoreList(List<Score> scoreList) {
+        this.scoreList = scoreList;
+    }
+}

dhp-workflows/dhp-aggregation/src/main/java/eu/dnetlib/dhp/actionmanager/bipfinder/KeyValue.java

@@ -0,0 +1,25 @@
+package eu.dnetlib.dhp.actionmanager.bipfinder;
+
+import java.io.Serializable;
+
+public class KeyValue implements Serializable {
+
+    private String key;
+    private String value;
+
+    public String getKey() {
+        return key;
+    }
+
+    public void setKey(String key) {
+        this.key = key;
+    }
+
+    public String getValue() {
+        return value;
+    }
+
+    public void setValue(String value) {
+        this.value = value;
+    }
+}

dhp-workflows/dhp-aggregation/src/main/java/eu/dnetlib/dhp/actionmanager/bipfinder/PreparedResult.java

@@ -0,0 +1,24 @@
+package eu.dnetlib.dhp.actionmanager.bipfinder;
+
+import java.io.Serializable;
+
+public class PreparedResult implements Serializable {
+    private String id; //openaire id
+    private String value; //doi
+
+    public String getId() {
+        return id;
+    }
+
+    public void setId(String id) {
+        this.id = id;
+    }
+
+    public String getValue() {
+        return value;
+    }
+
+    public void setValue(String value) {
+        this.value = value;
+    }
+}

dhp-workflows/dhp-aggregation/src/main/java/eu/dnetlib/dhp/actionmanager/bipfinder/Score.java

@@ -0,0 +1,26 @@
+package eu.dnetlib.dhp.actionmanager.bipfinder;
+
+import java.io.Serializable;
+import java.util.List;
+
+public class Score implements Serializable {
+
+    private String id;
+    private List<KeyValue> unit;
+
+    public String getId() {
+        return id;
+    }
+
+    public void setId(String id) {
+        this.id = id;
+    }
+
+    public List<KeyValue> getUnit() {
+        return unit;
+    }
+
+    public void setUnit(List<KeyValue> unit) {
+        this.unit = unit;
+    }
+}

dhp-workflows/dhp-aggregation/src/main/java/eu/dnetlib/dhp/actionmanager/bipfinder/SparkAtomicActionScoreJob.java

@@ -0,0 +1,203 @@
+package eu.dnetlib.dhp.actionmanager.bipfinder;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import eu.dnetlib.dhp.application.ArgumentApplicationParser;
+import eu.dnetlib.dhp.common.HdfsSupport;
+import eu.dnetlib.dhp.schema.action.AtomicAction;
+import eu.dnetlib.dhp.schema.oaf.*;
+import eu.dnetlib.dhp.schema.oaf.KeyValue;
+import org.apache.commons.io.IOUtils;
+import org.apache.hadoop.hdfs.qjournal.protocol.QJournalProtocolProtos;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapred.SequenceFileOutputFormat;
+import org.apache.spark.SparkConf;
+import org.apache.spark.api.java.JavaRDD;
+import org.apache.spark.api.java.JavaSparkContext;
+import org.apache.spark.api.java.function.FlatMapFunction;
+import org.apache.spark.api.java.function.MapFunction;
+import org.apache.spark.api.java.function.MapGroupsFunction;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Encoders;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.SparkSession;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import scala.Tuple2;
+
+import java.io.Serializable;
+import java.util.List;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+import static eu.dnetlib.dhp.common.SparkSessionSupport.runWithSparkSession;
+
+public class SparkAtomicActionScoreJob implements Serializable {
+
+    private static String DOI = "doi";
+    private static final Logger log = LoggerFactory.getLogger(SparkAtomicActionScoreJob.class);
+    private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
+    public static <I extends Result> void main(String[] args) throws Exception {
+
+        String jsonConfiguration = IOUtils
+                .toString(
+                        SparkAtomicActionScoreJob.class
+                                .getResourceAsStream(
+                                        "/eu/dnetlib/dhp/actionmanager/bipfinder/input_parameters.json"));
+
+        final ArgumentApplicationParser parser = new ArgumentApplicationParser(jsonConfiguration);
+
+        parser.parseArgument(args);
+
+        Boolean isSparkSessionManaged = Optional
+                .ofNullable(parser.get("isSparkSessionManaged"))
+                .map(Boolean::valueOf)
+                .orElse(Boolean.TRUE);
+
+        log.info("isSparkSessionManaged: {}", isSparkSessionManaged);
+
+        final String inputPath = parser.get("inputPath");
+        log.info("inputPath {}: ", inputPath);
+
+        final String outputPath = parser.get("outputPath");
+        log.info("outputPath {}: ", outputPath);
+
+        final String bipScorePath = parser.get("bipScorePath");
+        log.info("bipScorePath: {}", bipScorePath);
+
+        final String resultClassName = parser.get("resultTableName");
+        log.info("resultTableName: {}", resultClassName);
+
+        Class<I> inputClazz = (Class<I>) Class.forName(resultClassName);
+
+        SparkConf conf = new SparkConf();
+
+        runWithSparkSession(
+                conf,
+                isSparkSessionManaged,
+                spark -> {
+                    removeOutputDir(spark, outputPath);
+                    prepareResults(spark, inputPath, outputPath, bipScorePath, inputClazz);
+                });
+    }
+
+    private static <I extends Result> void prepareResults(SparkSession spark, String inputPath, String outputPath,
+                                                         String bipScorePath, Class<I> inputClazz) {
+
+        final JavaSparkContext sc = new JavaSparkContext(spark.sparkContext());
+
+        JavaRDD<BipDeserialize> bipDeserializeJavaRDD = sc.textFile(bipScorePath)
+                .map(item -> OBJECT_MAPPER.readValue(item, BipDeserialize.class));
+
+
+        Dataset<BipScore> bipScores = spark.createDataset(bipDeserializeJavaRDD.flatMap(entry ->
+                entry.keySet().stream().map(key -> {
+                    BipScore bs = new BipScore();
+                    bs.setId(key);
+                    bs.setScoreList(entry.get(key));
+                    return bs;
+                }).collect(Collectors.toList()).iterator()
+        ).rdd(), Encoders.bean(BipScore.class));
+
+        System.out.println(bipScores.count());
+
+        Dataset<I> results = readPath(spark, inputPath, inputClazz);
+
+        results.createOrReplaceTempView("result");
+
+        Dataset<PreparedResult> preparedResult = spark.sql("select pIde.value value, id " +
+                "from result " +
+                "lateral view explode (pid) p as pIde " +
+                "where dataInfo.deletedbyinference = false and pIde.qualifier.classid = '" + DOI + "'")
+                .as(Encoders.bean(PreparedResult.class));
+
+        Dataset<BipScore> tmp = bipScores.joinWith(preparedResult, bipScores.col("id").equalTo(preparedResult.col("value")),
+                "inner")
+                .map((MapFunction<Tuple2<BipScore, PreparedResult>, BipScore>) value -> {
+                    BipScore ret = value._1();
+                    ret.setId(value._2().getId());
+                    return ret;
+                }, Encoders.bean(BipScore.class));
+
+        tmp.groupByKey((MapFunction<BipScore, String>) value -> value.getId(), Encoders.STRING())
+        .mapGroups((MapGroupsFunction<String, BipScore, I>) (k, it) ->
+        {
+            Result ret = inputClazz.newInstance();
+            BipScore first = it.next();
+            ret.setId(first.getId());
+
+            ret.setMeasures(getMeasure(first));
+            it.forEachRemaining(value -> ret.getMeasures().addAll(getMeasure(value)));
+
+            return (I)ret;
+        }, Encoders.bean(inputClazz))
+                .toJavaRDD()
+                .map(p -> new AtomicAction(inputClazz, p))
+                .mapToPair(
+                        aa -> new Tuple2<>(new Text(aa.getClazz().getCanonicalName()),
+                                new Text(OBJECT_MAPPER.writeValueAsString(aa))))
+                .saveAsHadoopFile(outputPath, Text.class, Text.class, SequenceFileOutputFormat.class);
+        ;
+    }
+
+    public static Dataset<BipScore> getBipScoreDataset(Dataset<BipDeserialize> bipdeserialized){
+        return  bipdeserialized.flatMap((FlatMapFunction<BipDeserialize, BipScore>) bip ->
+                        bip.keySet().stream().map(key -> {
+                            BipScore bs = new BipScore();
+                            bs.setId(key);
+                            bs.setScoreList(bip.get(key));
+                            return bs;
+                        }).collect(Collectors.toList()).iterator()
+                , Encoders.bean(BipScore.class));
+    }
+
+    private static List<Measure> getMeasure(BipScore value) {
+        return value.getScoreList()
+                .stream()
+                .map(score -> {
+                    Measure m = new Measure();
+                    m.setId(score.getId());
+                    m.setUnit(score.getUnit().stream()
+                            .map(unit -> {
+                                KeyValue kv = new KeyValue();
+                                kv.setValue(unit.getValue());
+                                kv.setKey(unit.getKey());
+                                kv.setDataInfo(getDataInfo());
+                                return kv;
+                            }).collect(Collectors.toList()));
+                    return m;
+                }).collect(Collectors.toList());
+    }
+
+
+
+    private static DataInfo getDataInfo() {
+        DataInfo di = new DataInfo();
+        di.setInferred(false);
+        di.setInvisible(false);
+        di.setDeletedbyinference(false);
+        di.setTrust("");
+        Qualifier qualifier = new Qualifier();
+        qualifier.setClassid("sysimport:actionset");
+        qualifier.setClassname("Harvested");
+        qualifier.setSchemename("dnet:provenanceActions");
+        qualifier.setSchemeid("dnet:provenanceActions");
+        di.setProvenanceaction(qualifier);
+        return di;
+    }
+
+
+    private static void removeOutputDir(SparkSession spark, String path) {
+        HdfsSupport.remove(path, spark.sparkContext().hadoopConfiguration());
+    }
+
+    public static <R> Dataset<R> readPath(
+            SparkSession spark, String inputPath, Class<R> clazz) {
+        return spark
+                .read()
+                .textFile(inputPath)
+                .map((MapFunction<String, R>) value -> OBJECT_MAPPER.readValue(value, clazz), Encoders.bean(clazz));
+    }
+
+
+}

dhp-workflows/dhp-aggregation/src/main/resources/eu/dnetlib/dhp/actionmanager/bipfinder/input_parameters.json

@@ -0,0 +1,32 @@
+[
+  {
+  "paramName": "issm",
+  "paramLongName": "isSparkSessionManaged",
+  "paramDescription": "when true will stop SparkSession after job execution",
+  "paramRequired": false
+},
+{
+"paramName": "ip",
+"paramLongName": "inputPath",
+"paramDescription": "the URL from where to get the programme file",
+"paramRequired": true
+},
+{
+"paramName": "o",
+"paramLongName": "outputPath",
+"paramDescription": "the path of the new ActionSet",
+"paramRequired": true
+},
+  {
+    "paramName": "rtn",
+    "paramLongName": "resultTableName",
+    "paramDescription": "the path of the new ActionSet",
+    "paramRequired": true
+  },
+  {
+    "paramName": "bsp",
+    "paramLongName": "bipScorePath",
+    "paramDescription": "the path of the new ActionSet",
+    "paramRequired": true
+  }
+]

dhp-workflows/dhp-aggregation/src/test/java/eu/dnetlib/dhp/actionmanager/bipfinder/SparkAtomicActionScoreJobTest.java

@@ -0,0 +1,160 @@
+package eu.dnetlib.dhp.actionmanager.bipfinder;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import eu.dnetlib.dhp.schema.action.AtomicAction;
+import eu.dnetlib.dhp.schema.oaf.*;
+import eu.dnetlib.dhp.schema.oaf.KeyValue;
+import org.apache.commons.io.FileUtils;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapred.SequenceFileOutputFormat;
+import org.apache.spark.SparkConf;
+import org.apache.spark.api.java.JavaRDD;
+import org.apache.spark.api.java.JavaSparkContext;
+import org.apache.spark.api.java.function.FilterFunction;
+import org.apache.spark.api.java.function.ForeachFunction;
+import org.apache.spark.api.java.function.MapFunction;
+import org.apache.spark.api.java.function.MapGroupsFunction;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Encoders;
+import org.apache.spark.sql.SparkSession;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import scala.Tuple2;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.stream.Collectors;
+
+public class SparkAtomicActionScoreJobTest {
+
+    private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
+    private static SparkSession spark;
+
+    private static Path workingDir;
+    private static final Logger log = LoggerFactory
+            .getLogger(SparkAtomicActionScoreJobTest.class);
+
+    @BeforeAll
+    public static void beforeAll() throws IOException {
+        workingDir = Files
+                .createTempDirectory(SparkAtomicActionScoreJobTest.class.getSimpleName());
+        log.info("using work dir {}", workingDir);
+
+        SparkConf conf = new SparkConf();
+        conf.setAppName(SparkAtomicActionScoreJobTest.class.getSimpleName());
+
+        conf.setMaster("local[*]");
+        conf.set("spark.driver.host", "localhost");
+        conf.set("hive.metastore.local", "true");
+        conf.set("spark.ui.enabled", "false");
+        conf.set("spark.sql.warehouse.dir", workingDir.toString());
+        conf.set("hive.metastore.warehouse.dir", workingDir.resolve("warehouse").toString());
+
+        spark = SparkSession
+                .builder()
+                .appName(SparkAtomicActionScoreJobTest.class.getSimpleName())
+                .config(conf)
+                .getOrCreate();
+    }
+
+    @AfterAll
+    public static void afterAll() throws IOException {
+        FileUtils.deleteDirectory(workingDir.toFile());
+        spark.stop();
+    }
+
+    @Test
+    public <I extends Result> void numberDistinctProjectTest() throws Exception {
+        String bipScoresPath =  getClass().getResource("/eu/dnetlib/dhp/actionmanager/bipfinder/bip_scores.json").getPath();
+        String inputPath = getClass()
+                .getResource(
+                        "/eu/dnetlib/dhp/actionmanager/bipfinder/publication.json")
+                .getPath();
+
+        SparkAtomicActionScoreJob.main(
+                        new String[] {
+                                "-isSparkSessionManaged",
+                                Boolean.FALSE.toString(),
+                                "-inputPath",
+                                inputPath,
+                                "-bipScorePath",
+                               bipScoresPath,
+                                "-resultTableName",
+                                "eu.dnetlib.dhp.schema.oaf.Publication",
+                                "-outputPath",
+                                workingDir.toString() + "/actionSet"
+                        });
+
+
+        final JavaSparkContext sc = new JavaSparkContext(spark.sparkContext());
+
+        JavaRDD<Publication> tmp = sc
+                .sequenceFile(workingDir.toString() + "/actionSet", Text.class, Text.class)
+                .map(value -> OBJECT_MAPPER.readValue(value._2().toString(), AtomicAction.class))
+                .map(aa -> ((Publication) aa.getPayload()));
+
+        Assertions.assertTrue(tmp.count() == 1);
+
+//        Dataset<Publication> verificationDataset = spark.createDataset(tmp.rdd(), Encoders.bean(Publication.class));
+//        verificationDataset.createOrReplaceTempView("project");
+
+//        Dataset<Row> execverification = spark
+//                .sql(
+//                        "SELECT id, class classification, h2020topiccode, h2020topicdescription FROM project LATERAL VIEW EXPLODE(h2020classification) c as class ");
+//
+//        Assertions
+//                .assertEquals(
+//                        "H2020-EU.3.4.7.",
+//                        execverification
+//                                .filter("id = '40|corda__h2020::2c7298913008865ba784e5c1350a0aa5'")
+//                                .select("classification.h2020Programme.code")
+//                                .collectAsList()
+//                                .get(0)
+//                                .getString(0));
+
+
+
+
+    }
+
+    private static List<Measure> getMeasure(BipScore value) {
+        return value.getScoreList()
+                .stream()
+                .map(score -> {
+                    Measure m = new Measure();
+                    m.setId(score.getId());
+                    m.setUnit(score.getUnit().stream()
+                            .map(unit -> {
+                                eu.dnetlib.dhp.schema.oaf.KeyValue kv = new KeyValue();
+                                kv.setValue(unit.getValue());
+                                kv.setKey(unit.getKey());
+                                kv.setDataInfo(getDataInfo());
+                                return kv;
+                            }).collect(Collectors.toList()));
+                    return m;
+                }).collect(Collectors.toList());
+    }
+
+
+    private static DataInfo getDataInfo() {
+        DataInfo di = new DataInfo();
+        di.setInferred(false);
+        di.setInvisible(false);
+        di.setDeletedbyinference(false);
+        di.setTrust("");
+        Qualifier qualifier = new Qualifier();
+        qualifier.setClassid("sysimport:actionset");
+        qualifier.setClassname("Harvested");
+        qualifier.setSchemename("dnet:provenanceActions");
+        qualifier.setSchemeid("dnet:provenanceActions");
+        di.setProvenanceaction(qualifier);
+        return di;
+    }
+}