new parametrized implementation for country propagation

dhp-workflows/dhp-propagation/src/main/java/eu/dnetlib/dhp/countrypropagation/Country.java

@@ -0,0 +1,24 @@
+package eu.dnetlib.dhp.countrypropagation;
+
+import java.io.Serializable;
+
+public class Country implements Serializable {
+    private String classid;
+    private String classname;
+
+    public String getClassid() {
+        return classid;
+    }
+
+    public void setClassid(String classid) {
+        this.classid = classid;
+    }
+
+    public String getClassname() {
+        return classname;
+    }
+
+    public void setClassname(String classname) {
+        this.classname = classname;
+    }
+}

dhp-workflows/dhp-propagation/src/main/java/eu/dnetlib/dhp/countrypropagation/DatasourceCountry.java

@@ -4,7 +4,7 @@ import java.io.Serializable;
 
 public class DatasourceCountry implements Serializable {
     private String dataSourceId;
-    private String country;
+    private Country country;
 
     public String getDataSourceId() {
         return dataSourceId;
@@ -14,11 +14,11 @@ public class DatasourceCountry implements Serializable {
         this.dataSourceId = dataSourceId;
     }
 
-    public String getCountry() {
+    public Country getCountry() {
         return country;
     }
 
-    public void setCountry(String country) {
+    public void setCountry(Country country) {
         this.country = country;
     }
 }

dhp-workflows/dhp-propagation/src/main/java/eu/dnetlib/dhp/countrypropagation/PrepareDatasourceCountryAssociation.java

@@ -29,14 +29,14 @@ import org.slf4j.LoggerFactory;
  * relation provides between Datasource and Organization is exploited to get the country for the datasource
  */
 
-public class PrepareResultCountryAssociation {
+public class PrepareDatasourceCountryAssociation {
-    private static final Logger log = LoggerFactory.getLogger(PrepareResultCountryAssociation.class);
+    private static final Logger log = LoggerFactory.getLogger(PrepareDatasourceCountryAssociation.class);
 
     private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
 
     public static void main(String[] args) throws Exception {
 
-        String jsonConfiguration = IOUtils.toString(PrepareResultCountryAssociation.class
+        String jsonConfiguration = IOUtils.toString(PrepareDatasourceCountryAssociation.class
                 .getResourceAsStream("/eu/dnetlib/dhp/countrypropagation/input_prepare_dc_assoc.json"));
 
         final ArgumentApplicationParser parser = new ArgumentApplicationParser(
@@ -97,7 +97,7 @@ public class PrepareResultCountryAssociation {
         relation.createOrReplaceTempView("relation");
         organization.createOrReplaceTempView("organization");
 
-        String query = "SELECT source ds, country.classid country " +
+        String query = "SELECT source ds, named_struct('classid', country.classid, 'classname', country.classname) country " +
                 "FROM ( SELECT id " +
                 "       FROM datasource " +
                 "       WHERE (datainfo.deletedbyinference = false " + whitelisted + ") " +

dhp-workflows/dhp-propagation/src/main/java/eu/dnetlib/dhp/countrypropagation/ResultCountrySet.java

@@ -0,0 +1,25 @@
+package eu.dnetlib.dhp.countrypropagation;
+
+import java.io.Serializable;
+import java.util.Set;
+
+public class ResultCountrySet implements Serializable {
+    private String resultId;
+    private Set<Country> countrySet;
+
+    public String getResultId() {
+        return resultId;
+    }
+
+    public void setResultId(String resultId) {
+        this.resultId = resultId;
+    }
+
+    public Set<Country> getCountrySet() {
+        return countrySet;
+    }
+
+    public void setCountrySet(Set<Country> countrySet) {
+        this.countrySet = countrySet;
+    }
+}

dhp-workflows/dhp-propagation/src/main/java/eu/dnetlib/dhp/countrypropagation/SparkCountryPropagationJob.java

@@ -100,7 +100,7 @@ public class SparkCountryPropagationJob {
         datasource.createOrReplaceTempView("datasource");
         relation.createOrReplaceTempView("relation");
         organization.createOrReplaceTempView("organization");
-
+//todo add length(country.classid)>0
         String query = "SELECT source ds, target org, country.classid country " +
                 "FROM ( SELECT id " +
                 "FROM datasource " +
@@ -118,6 +118,7 @@ public class SparkCountryPropagationJob {
                 "WHERE datainfo.deletedbyinference = false ) o " +
                 "ON o.id = rel.target";
 
+        //todo broadcast
         Dataset<Row> rels = spark.sql(query);
         rels.createOrReplaceTempView("rels");
 
@@ -268,11 +269,11 @@ public class SparkCountryPropagationJob {
                 });
     }
 
-    private static JavaRDD<Row> propagateOnResult(SparkSession spark, String table) {
+    private static JavaRDD<Row> propagateOnResult(SparkSession spark, String result_type) {
         String query;
         query = "SELECT id, inst.collectedfrom.key cf , inst.hostedby.key hb " +
                 "FROM ( SELECT id, instance " +
-                "FROM " + table +
+                "FROM " + result_type +
                 " WHERE datainfo.deletedbyinference = false)  ds " +
                 "LATERAL VIEW EXPLODE(instance) i AS inst";
         Dataset<Row> cfhb = spark.sql(query);

dhp-workflows/dhp-propagation/src/main/java/eu/dnetlib/dhp/countrypropagation/SparkCountryPropagationJob2.java

@@ -5,7 +5,6 @@ import eu.dnetlib.dhp.application.ArgumentApplicationParser;
 import eu.dnetlib.dhp.schema.oaf.*;
 import org.apache.commons.io.IOUtils;
 import org.apache.spark.SparkConf;
-import org.apache.spark.api.java.JavaPairRDD;
 import org.apache.spark.api.java.JavaRDD;
 import org.apache.spark.api.java.JavaSparkContext;
 import org.apache.spark.api.java.function.MapFunction;
@@ -41,7 +40,7 @@ public class SparkCountryPropagationJob2 {
 
         final String outputPath = "/tmp/provision/propagation/countrytoresultfrominstitutionalrepositories";
         final String datasourcecountrypath = outputPath + "/prepared_datasource_country";
-        final String resultClassName = parser.get("resultClazz");
+        final String resultClassName = parser.get("resultTableName");
 
         final String resultType = resultClassName.substring(resultClassName.lastIndexOf(".")+1);
 
@@ -67,7 +66,8 @@ public class SparkCountryPropagationJob2 {
         //broadcasting the result of the preparation step
         Broadcast<Dataset<DatasourceCountry>> broadcast_datasourcecountryassoc = sc.broadcast(datasourcecountryassoc);
 
-        Dataset<Row> potentialUpdates = getPotentialResultToUpdate(spark, inputPath, resultClazz, broadcast_datasourcecountryassoc);
+        Dataset<ResultCountrySet> potentialUpdates = getPotentialResultToUpdate(spark, inputPath, resultClazz,
+                broadcast_datasourcecountryassoc).as(Encoders.bean(ResultCountrySet.class));
 
         if(writeUpdates){
             writeUpdates(potentialUpdates.toJavaRDD(), outputPath + "/update_" + resultType);
@@ -80,7 +80,7 @@ public class SparkCountryPropagationJob2 {
 
     }
 
-    private static <R extends Result> void updateResultTable(SparkSession spark, Dataset<Row> potentialUpdates,
+    private static <R extends Result> void updateResultTable(SparkSession spark, Dataset<ResultCountrySet> potentialUpdates,
                                                              String inputPath,
                                                              Class<R> resultClazz,
                                                              String outputPath) {
@@ -91,23 +91,24 @@ public class SparkCountryPropagationJob2 {
                 .map(r -> new Tuple2<>(r.getId(), r),
                 Encoders.tuple(Encoders.STRING(), Encoders.bean(resultClazz)));
 
-        Dataset<Tuple2<String, List>> potential_update_pair = potentialUpdates.map(pu -> new Tuple2<>(pu.getString(0), pu.getList(1)),
+        Dataset<Tuple2<String, ResultCountrySet>> potential_update_pair = potentialUpdates.map(pu -> new Tuple2<>(pu.getResultId(),
-                Encoders.tuple(Encoders.STRING(), Encoders.bean(List.class)));
+                        pu),
+                Encoders.tuple(Encoders.STRING(), Encoders.bean(ResultCountrySet.class)));
 
         Dataset<R> new_table = result_pair
                 .joinWith(potential_update_pair, result_pair.col("_1").equalTo(potential_update_pair.col("_1")), "left")
-                .map((MapFunction<Tuple2<Tuple2<String, R>, Tuple2<String, List>>, R>) value -> {
+                .map((MapFunction<Tuple2<Tuple2<String, R>, Tuple2<String, ResultCountrySet>>, R>) value -> {
                     R r = value._1()._2();
-                    Optional<List<Object>> potentialNewCountries = Optional.ofNullable(value._2()).map(Tuple2::_2);
+                    Optional<ResultCountrySet> potentialNewCountries = Optional.ofNullable(value._2()).map(Tuple2::_2);
-                    if (potentialNewCountries != null) {
+                    if (potentialNewCountries.isPresent()) {
                         HashSet<String> countries = new HashSet<>();
                         for (Qualifier country : r.getCountry()) {
                             countries.add(country.getClassid());
                         }
 
-                        for (Object country : potentialNewCountries.get()) {
+                        for (Country country : potentialNewCountries.get().getCountrySet()) {
-                            if (!countries.contains(country)) {
+                            if (!countries.contains(country.getClassid())) {
-                                r.getCountry().add(getCountry((String) country));
+                                r.getCountry().add(getCountry(country.getClassid(),country.getClassname()));
                             }
                         }
                     }
@@ -117,7 +118,7 @@ public class SparkCountryPropagationJob2 {
 
 
             log.info("Saving graph table to path: {}", outputPath);
-            result
+            new_table
                     .toJSON()
                     .write()
                     .option("compression", "gzip")
@@ -151,6 +152,7 @@ public class SparkCountryPropagationJob2 {
 
     private static Dataset<Row> countryPropagationAssoc(SparkSession spark,
                                                         Broadcast<Dataset<DatasourceCountry>> broadcast_datasourcecountryassoc){
+
         Dataset<DatasourceCountry> datasource_country = broadcast_datasourcecountryassoc.value();
         datasource_country.createOrReplaceTempView("datasource_country");
 
@@ -183,7 +185,7 @@ public class SparkCountryPropagationJob2 {
                 .as(Encoders.bean(DatasourceCountry.class));
     }
 
-    private static void writeUpdates(JavaRDD<Row> potentialUpdates, String outputPath){
+    private static void writeUpdates(JavaRDD<ResultCountrySet> potentialUpdates, String outputPath){
         potentialUpdates.map(u -> OBJECT_MAPPER.writeValueAsString(u))
                 .saveAsTextFile(outputPath);
     }