refactoring

dhp-workflows/dhp-propagation/src/main/java/eu/dnetlib/dhp/resulttocommunityfromorganization/PrepareResultCommunitySet.java

@@ -1,20 +1,19 @@
 package eu.dnetlib.dhp.resulttocommunityfromorganization;
 
+import static eu.dnetlib.dhp.PropagationConstant.*;
+import static eu.dnetlib.dhp.common.SparkSessionSupport.runWithSparkHiveSession;
+
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.gson.Gson;
 import eu.dnetlib.dhp.application.ArgumentApplicationParser;
 import eu.dnetlib.dhp.schema.oaf.Relation;
+import java.util.*;
 import org.apache.commons.io.IOUtils;
 import org.apache.spark.SparkConf;
 import org.apache.spark.sql.*;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.util.*;
-
-import static eu.dnetlib.dhp.PropagationConstant.*;
-import static eu.dnetlib.dhp.common.SparkSessionSupport.runWithSparkHiveSession;
-
 public class PrepareResultCommunitySet {
 
     private static final Logger log = LoggerFactory.getLogger(PrepareResultCommunitySet.class);
@@ -22,11 +21,12 @@ public class PrepareResultCommunitySet {
     private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
 
     public static void main(String[] args) throws Exception {
-        String jsonConfiguration = IOUtils.toString(PrepareResultCommunitySet.class
+        String jsonConfiguration =
-                .getResourceAsStream("/eu/dnetlib/dhp/resulttocommunityfromorganization/input_preparecommunitytoresult_parameters.json"));
+                IOUtils.toString(
+                        PrepareResultCommunitySet.class.getResourceAsStream(
+                                "/eu/dnetlib/dhp/resulttocommunityfromorganization/input_preparecommunitytoresult_parameters.json"));
 
-        final ArgumentApplicationParser parser = new ArgumentApplicationParser(
+        final ArgumentApplicationParser parser = new ArgumentApplicationParser(jsonConfiguration);
-                jsonConfiguration);
 
         parser.parseArgument(args);
 
@@ -39,13 +39,19 @@ public class PrepareResultCommunitySet {
         final String outputPath = parser.get("outputPath");
         log.info("outputPath: {}", outputPath);
 
-        final OrganizationMap organizationMap = new Gson().fromJson(parser.get("organizationtoresultcommunitymap"), OrganizationMap.class);
+        final OrganizationMap organizationMap =
+                new Gson()
+                        .fromJson(
+                                parser.get("organizationtoresultcommunitymap"),
+                                OrganizationMap.class);
         log.info("organizationMap: {}", new Gson().toJson(organizationMap));
 
         SparkConf conf = new SparkConf();
         conf.set("hive.metastore.uris", parser.get("hive_metastore_uris"));
 
-        runWithSparkHiveSession(conf, isSparkSessionManaged,
+        runWithSparkHiveSession(
+                conf,
+                isSparkSessionManaged,
                 spark -> {
                     if (isTest(parser)) {
                         removeOutputDir(spark, outputPath);
@@ -54,43 +60,54 @@ public class PrepareResultCommunitySet {
                 });
     }
 
-    private static void prepareInfo(SparkSession spark, String inputPath, String outputPath, OrganizationMap organizationMap) {
+    private static void prepareInfo(
+            SparkSession spark,
+            String inputPath,
+            String outputPath,
+            OrganizationMap organizationMap) {
         Dataset<Relation> relation = readRelations(spark, inputPath);
         relation.createOrReplaceTempView("relation");
 
-        String query = "SELECT result_organization.source resultId, result_organization.target orgId, org_set merges " +
+        String query =
-                "FROM (SELECT source, target " +
+                "SELECT result_organization.source resultId, result_organization.target orgId, org_set merges "
-                "      FROM relation " +
+                        + "FROM (SELECT source, target "
-                "      WHERE datainfo.deletedbyinference = false " +
+                        + "      FROM relation "
-                "      AND relClass = '" + RELATION_RESULT_ORGANIZATION_REL_CLASS + "') result_organization " +
+                        + "      WHERE datainfo.deletedbyinference = false "
-                "LEFT JOIN (SELECT source, collect_set(target) org_set " +
+                        + "      AND relClass = '"
-                "      FROM relation " +
+                        + RELATION_RESULT_ORGANIZATION_REL_CLASS
-                "      WHERE datainfo.deletedbyinference = false " +
+                        + "') result_organization "
-                "      AND relClass = '" + RELATION_REPRESENTATIVERESULT_RESULT_CLASS + "' " +
+                        + "LEFT JOIN (SELECT source, collect_set(target) org_set "
-                "      GROUP BY source) organization_organization " +
+                        + "      FROM relation "
-                "ON result_organization.target = organization_organization.source ";
+                        + "      WHERE datainfo.deletedbyinference = false "
+                        + "      AND relClass = '"
+                        + RELATION_REPRESENTATIVERESULT_RESULT_CLASS
+                        + "' "
+                        + "      GROUP BY source) organization_organization "
+                        + "ON result_organization.target = organization_organization.source ";
 
-        org.apache.spark.sql.Dataset<ResultOrganizations> result_organizationset = spark.sql(query)
+        org.apache.spark.sql.Dataset<ResultOrganizations> result_organizationset =
-                .as(Encoders.bean(ResultOrganizations.class));
+                spark.sql(query).as(Encoders.bean(ResultOrganizations.class));
 
         result_organizationset
-                .map(value -> {
+                .map(
+                        value -> {
                             String rId = value.getResultId();
-                    List<String> orgs = value.getMerges();
+                            Optional<List<String>> orgs = Optional.ofNullable(value.getMerges());
                             String oTarget = value.getOrgId();
                             Set<String> communitySet = new HashSet<>();
                             if (organizationMap.containsKey(oTarget)) {
                                 communitySet.addAll(organizationMap.get(oTarget));
                             }
-                    try{
+                            if (orgs.isPresent())
-                        for (String oId : orgs) {
+                                // try{
+                                for (String oId : orgs.get()) {
                                     if (organizationMap.containsKey(oId)) {
                                         communitySet.addAll(organizationMap.get(oId));
                                     }
                                 }
-                    }catch(Exception e){
+                            //                    }catch(Exception e){
-
+                            //
-                    }
+                            //                    }
                             if (communitySet.size() > 0) {
                                 ResultCommunityList rcl = new ResultCommunityList();
                                 rcl.setResultId(rId);
@@ -100,7 +117,8 @@ public class PrepareResultCommunitySet {
                                 return rcl;
                             }
                             return null;
-                }, Encoders.bean(ResultCommunityList.class))
+                        },
+                        Encoders.bean(ResultCommunityList.class))
                 .filter(r -> r != null)
                 .toJSON()
                 .write()
@@ -108,7 +126,4 @@ public class PrepareResultCommunitySet {
                 .option("compression", "gzip")
                 .text(outputPath);
     }
-
-
-
 }

dhp-workflows/dhp-propagation/src/main/java/eu/dnetlib/dhp/resulttocommunityfromorganization/SparkResultToCommunityFromOrganizationJob2.java

@@ -1,8 +1,13 @@
 package eu.dnetlib.dhp.resulttocommunityfromorganization;
 
+import static eu.dnetlib.dhp.PropagationConstant.*;
+import static eu.dnetlib.dhp.common.SparkSessionSupport.runWithSparkHiveSession;
+
 import com.fasterxml.jackson.databind.ObjectMapper;
 import eu.dnetlib.dhp.application.ArgumentApplicationParser;
 import eu.dnetlib.dhp.schema.oaf.*;
+import java.util.*;
+import java.util.stream.Collectors;
 import org.apache.commons.io.IOUtils;
 import org.apache.spark.SparkConf;
 import org.apache.spark.sql.Encoders;
@@ -10,25 +15,21 @@ import org.apache.spark.sql.SaveMode;
 import org.apache.spark.sql.SparkSession;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import java.util.*;
-import java.util.stream.Collectors;
-
-import static eu.dnetlib.dhp.PropagationConstant.*;
-import static eu.dnetlib.dhp.common.SparkSessionSupport.runWithSparkHiveSession;
 
 public class SparkResultToCommunityFromOrganizationJob2 {
 
-    private static final Logger log = LoggerFactory.getLogger(SparkResultToCommunityFromOrganizationJob2.class);
+    private static final Logger log =
+            LoggerFactory.getLogger(SparkResultToCommunityFromOrganizationJob2.class);
 
     private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
 
-
     public static void main(String[] args) throws Exception {
-        String jsonConfiguration = IOUtils.toString(SparkResultToCommunityFromOrganizationJob2.class
+        String jsonConfiguration =
-                .getResourceAsStream("/eu/dnetlib/dhp/resulttocommunityfromorganization/input_communitytoresult_parameters.json"));
+                IOUtils.toString(
+                        SparkResultToCommunityFromOrganizationJob2.class.getResourceAsStream(
+                                "/eu/dnetlib/dhp/resulttocommunityfromorganization/input_communitytoresult_parameters.json"));
 
-        final ArgumentApplicationParser parser = new ArgumentApplicationParser(
+        final ArgumentApplicationParser parser = new ArgumentApplicationParser(jsonConfiguration);
-                jsonConfiguration);
 
         parser.parseArgument(args);
 
@@ -47,43 +48,53 @@ public class SparkResultToCommunityFromOrganizationJob2 {
         final String resultClassName = parser.get("resultTableName");
         log.info("resultTableName: {}", resultClassName);
 
-        final Boolean saveGraph = Optional
+        final Boolean saveGraph =
-                .ofNullable(parser.get("saveGraph"))
+                Optional.ofNullable(parser.get("saveGraph"))
                         .map(Boolean::valueOf)
                         .orElse(Boolean.TRUE);
         log.info("saveGraph: {}", saveGraph);
 
-        Class<? extends Result> resultClazz = (Class<? extends Result>) Class.forName(resultClassName);
+        Class<? extends Result> resultClazz =
+                (Class<? extends Result>) Class.forName(resultClassName);
 
         SparkConf conf = new SparkConf();
         conf.set("hive.metastore.uris", parser.get("hive_metastore_uris"));
 
-
+        runWithSparkHiveSession(
-        runWithSparkHiveSession(conf, isSparkSessionManaged,
+                conf,
+                isSparkSessionManaged,
                 spark -> {
                     if (isTest(parser)) {
                         removeOutputDir(spark, outputPath);
                     }
                     execPropagation(spark, inputPath, outputPath, resultClazz, possibleupdatespath);
                 });
-
     }
 
-    private static <R extends Result> void execPropagation(SparkSession spark, String inputPath, String outputPath,
+    private static <R extends Result> void execPropagation(
-                                                           Class<R> resultClazz, String possibleUpdatesPath) {
+            SparkSession spark,
-        org.apache.spark.sql.Dataset<ResultCommunityList> possibleUpdates = readResultCommunityList(spark, possibleUpdatesPath);
+            String inputPath,
+            String outputPath,
+            Class<R> resultClazz,
+            String possibleUpdatesPath) {
+        org.apache.spark.sql.Dataset<ResultCommunityList> possibleUpdates =
+                readResultCommunityList(spark, possibleUpdatesPath);
         org.apache.spark.sql.Dataset<R> result = readPathEntity(spark, inputPath, resultClazz);
 
-
+        result.joinWith(
-        result
+                        possibleUpdates,
-                .joinWith(possibleUpdates, result.col("id").equalTo(possibleUpdates.col("resultId")),
+                        result.col("id").equalTo(possibleUpdates.col("resultId")),
                         "left_outer")
-                .map(value -> {
+                .map(
+                        value -> {
                             R ret = value._1();
                             Optional<ResultCommunityList> rcl = Optional.ofNullable(value._2());
                             if (rcl.isPresent()) {
                                 ArrayList<String> communitySet = rcl.get().getCommunityList();
-                        List<String> contextList = ret.getContext().stream().map(con -> con.getId()).collect(Collectors.toList());
+                                List<String> contextList =
+                                        ret.getContext().stream()
+                                                .map(con -> con.getId())
+                                                .collect(Collectors.toList());
                                 Result res = new Result();
                                 res.setId(ret.getId());
                                 List<Context> propagatedContexts = new ArrayList<>();
@@ -91,8 +102,12 @@ public class SparkResultToCommunityFromOrganizationJob2 {
                                     if (!contextList.contains(cId)) {
                                         Context newContext = new Context();
                                         newContext.setId(cId);
-                                newContext.setDataInfo(Arrays.asList(getDataInfo(PROPAGATION_DATA_INFO_TYPE,
+                                        newContext.setDataInfo(
-                                        PROPAGATION_RESULT_COMMUNITY_ORGANIZATION_CLASS_ID, PROPAGATION_RESULT_COMMUNITY_ORGANIZATION_CLASS_NAME)));
+                                                Arrays.asList(
+                                                        getDataInfo(
+                                                                PROPAGATION_DATA_INFO_TYPE,
+                                                                PROPAGATION_RESULT_COMMUNITY_ORGANIZATION_CLASS_ID,
+                                                                PROPAGATION_RESULT_COMMUNITY_ORGANIZATION_CLASS_NAME)));
                                         propagatedContexts.add(newContext);
                                     }
                                 }
@@ -100,20 +115,12 @@ public class SparkResultToCommunityFromOrganizationJob2 {
                                 ret.mergeFrom(res);
                             }
                             return ret;
-                }, Encoders.bean(resultClazz))
+                        },
+                        Encoders.bean(resultClazz))
                 .toJSON()
                 .write()
                 .mode(SaveMode.Overwrite)
                 .option("compression", "gzip")
                 .text(outputPath);
-
     }
-
-    private static org.apache.spark.sql.Dataset<ResultCommunityList> readResultCommunityList(SparkSession spark, String possibleUpdatesPath) {
-        return spark
-                .read()
-                .textFile(possibleUpdatesPath)
-                .map(value -> OBJECT_MAPPER.readValue(value, ResultCommunityList.class), Encoders.bean(ResultCommunityList.class));
-    }
-
 }