minor changes

dhp-workflows/dhp-dedup-openaire/src/main/java/eu/dnetlib/dhp/oa/dedup/SparkCreateConnectedComponent.java

@@ -13,7 +13,6 @@ import eu.dnetlib.pace.util.MapDocumentUtil;
 import org.apache.commons.io.IOUtils;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
-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;
@@ -29,6 +28,7 @@ import scala.Tuple2;
 
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.Iterator;
 import java.util.List;
 
 public class SparkCreateConnectedComponent extends AbstractSparkAction {
@@ -45,7 +45,7 @@ public class SparkCreateConnectedComponent extends AbstractSparkAction {
                         SparkCreateSimRels.class.getResourceAsStream("/eu/dnetlib/dhp/oa/dedup/createCC_parameters.json")));
         parser.parseArgument(args);
 
-        new SparkCreateSimRels(parser, getSparkSession(parser)).run(ISLookupClientFactory.getLookUpService(parser.get("isLookUpUrl")));
+        new SparkCreateConnectedComponent(parser, getSparkSession(parser)).run(ISLookupClientFactory.getLookUpService(parser.get("isLookUpUrl")));
     }
 
     @Override
@@ -56,12 +56,16 @@ public class SparkCreateConnectedComponent extends AbstractSparkAction {
         final String isLookUpUrl = parser.get("isLookUpUrl");
         final String actionSetId = parser.get("actionSetId");
 
+        System.out.println(String.format("graphBasePath: '%s'", graphBasePath));
+        System.out.println(String.format("isLookUpUrl:   '%s'", isLookUpUrl));
+        System.out.println(String.format("actionSetId:   '%s'", actionSetId));
+        System.out.println(String.format("workingPath:   '%s'", workingPath));
+
         final JavaSparkContext sc = new JavaSparkContext(spark.sparkContext());
 
         for (DedupConfig dedupConf: getConfigurations(isLookUpService, actionSetId)) {
-
-            final String entity = dedupConf.getWf().getEntityType();
             final String subEntity = dedupConf.getWf().getSubEntityValue();
+            System.out.println(String.format("Creating mergerels for: '%s'", subEntity));
 
             final JavaPairRDD<Object, String> vertexes = sc.textFile(graphBasePath + "/" + subEntity)
                     .map(s -> MapDocumentUtil.getJPathString(dedupConf.getWf().getIdPath(), s))
@@ -72,28 +76,35 @@ public class SparkCreateConnectedComponent extends AbstractSparkAction {
             final Dataset<Relation> similarityRelations = spark.read().load(DedupUtility.createSimRelPath(workingPath, actionSetId, subEntity)).as(Encoders.bean(Relation.class));
             final RDD<Edge<String>> edgeRdd = similarityRelations.javaRDD().map(it -> new Edge<>(getHashcode(it.getSource()), getHashcode(it.getTarget()), it.getRelClass())).rdd();
             final JavaRDD<ConnectedComponent> cc = GraphProcessor.findCCs(vertexes.rdd(), edgeRdd, dedupConf.getWf().getMaxIterations()).toJavaRDD();
-            final Dataset<Relation> mergeRelation = spark.createDataset(cc.filter(k -> k.getDocIds().size() > 1).flatMap((FlatMapFunction<ConnectedComponent, Relation>) c ->
+            final Dataset<Relation> mergeRelation = spark.createDataset(cc.filter(k -> k.getDocIds().size() > 1)
-                    c.getDocIds()
+                    .flatMap(this::ccToMergeRel).rdd(), Encoders.bean(Relation.class));
-                            .stream()
+
-                            .flatMap(id -> {
+            mergeRelation
-                                List<Relation> tmp = new ArrayList<>();
+                    .write().mode("overwrite")
-                                Relation r = new Relation();
+                    .save(DedupUtility.createMergeRelPath(workingPath, actionSetId, subEntity));
-                                r.setSource(c.getCcId());
-                                r.setTarget(id);
-                                r.setRelClass("merges");
-                                tmp.add(r);
-                                r = new Relation();
-                                r.setTarget(c.getCcId());
-                                r.setSource(id);
-                                r.setRelClass("isMergedIn");
-                                tmp.add(r);
-                                return tmp.stream();
-                            }).iterator()).rdd(), Encoders.bean(Relation.class));
-                mergeRelation.write().mode("overwrite").save(DedupUtility.createMergeRelPath(workingPath, actionSetId, subEntity));
         }
 
     }
 
+    public Iterator<Relation> ccToMergeRel(ConnectedComponent cc){
+        return cc.getDocIds()
+                .stream()
+                .flatMap(id -> {
+                    List<Relation> tmp = new ArrayList<>();
+                    Relation r = new Relation();
+                    r.setSource(cc.getCcId());
+                    r.setTarget(id);
+                    r.setRelClass("merges");
+                    tmp.add(r);
+                    r = new Relation();
+                    r.setTarget(cc.getCcId());
+                    r.setSource(id);
+                    r.setRelClass("isMergedIn");
+                    tmp.add(r);
+                    return tmp.stream();
+                }).iterator();
+    }
+
     public  static long getHashcode(final String id) {
         return Hashing.murmur3_128().hashString(id).asLong();
     }

dhp-workflows/dhp-dedup-openaire/src/main/java/eu/dnetlib/dhp/oa/dedup/SparkCreateDedupRecord.java

@@ -32,7 +32,7 @@ public class SparkCreateDedupRecord extends AbstractSparkAction {
                         SparkCreateSimRels.class.getResourceAsStream("/eu/dnetlib/dhp/oa/dedup/createDedupRecord_parameters.json")));
         parser.parseArgument(args);
 
-        new SparkCreateSimRels(parser, getSparkSession(parser)).run(ISLookupClientFactory.getLookUpService(parser.get("isLookUpUrl")));
+        new SparkCreateDedupRecord(parser, getSparkSession(parser)).run(ISLookupClientFactory.getLookUpService(parser.get("isLookUpUrl")));
     }
 
     @Override
@@ -43,10 +43,16 @@ public class SparkCreateDedupRecord extends AbstractSparkAction {
         final String actionSetId = parser.get("actionSetId");
         final String workingPath = parser.get("workingPath");
 
+        System.out.println(String.format("graphBasePath: '%s'", graphBasePath));
+        System.out.println(String.format("isLookUpUrl:   '%s'", isLookUpUrl));
+        System.out.println(String.format("actionSetId:   '%s'", actionSetId));
+        System.out.println(String.format("workingPath:   '%s'", workingPath));
+
         final JavaSparkContext sc = new JavaSparkContext(spark.sparkContext());
 
         for (DedupConfig dedupConf: getConfigurations(isLookUpService, actionSetId)) {
             String subEntity = dedupConf.getWf().getSubEntityValue();
+            System.out.println(String.format("Creating deduprecords for: '%s'", subEntity));
 
             final String mergeRelPath = DedupUtility.createMergeRelPath(workingPath, actionSetId, subEntity);
             final String entityPath = DedupUtility.createEntityPath(graphBasePath, subEntity);

dhp-workflows/dhp-dedup-openaire/src/main/java/eu/dnetlib/dhp/oa/dedup/SparkCreateSimRels.java

@@ -61,16 +61,18 @@ public class SparkCreateSimRels extends AbstractSparkAction {
         final String workingPath = parser.get("workingPath");
 
         System.out.println(String.format("graphBasePath: '%s'", graphBasePath));
-        System.out.println(String.format("isLookUpUrl: '%s'", isLookUpUrl));
+        System.out.println(String.format("isLookUpUrl:   '%s'", isLookUpUrl));
-        System.out.println(String.format("actionSetId: '%s'", actionSetId));
+        System.out.println(String.format("actionSetId:   '%s'", actionSetId));
-        System.out.println(String.format("workingPath: '%s'", workingPath));
+        System.out.println(String.format("workingPath:   '%s'", workingPath));
 
         final JavaSparkContext sc = new JavaSparkContext(spark.sparkContext());
 
         //for each dedup configuration
         for (DedupConfig dedupConf: getConfigurations(isLookUpService, actionSetId)) {
+
             final String entity = dedupConf.getWf().getEntityType();
             final String subEntity = dedupConf.getWf().getSubEntityValue();
+            System.out.println(String.format("Creating simrels for: '%s'", subEntity));
 
             JavaPairRDD<String, MapDocument> mapDocument = sc.textFile(DedupUtility.createEntityPath(graphBasePath, subEntity))
                     .mapToPair((PairFunction<String, String, MapDocument>)  s -> {

dhp-workflows/dhp-dedup-openaire/src/main/java/eu/dnetlib/dhp/oa/dedup/SparkPropagateRelation.java

@@ -60,6 +60,10 @@ public class SparkPropagateRelation extends AbstractSparkAction {
         final String workingPath = parser.get("workingPath");
         final String dedupGraphPath = parser.get("dedupGraphPath");
 
+        System.out.println(String.format("graphBasePath: '%s'", graphBasePath));
+        System.out.println(String.format("workingPath:   '%s'", workingPath));
+        System.out.println(String.format("dedupGraphPath:'%s'", dedupGraphPath));
+
         final JavaSparkContext sc = new JavaSparkContext(spark.sparkContext());
 
         final Dataset<Relation> mergeRels = spark.read().load(DedupUtility.createMergeRelPath(workingPath, "*", "*")).as(Encoders.bean(Relation.class));

dhp-workflows/dhp-dedup-openaire/src/main/java/eu/dnetlib/dhp/oa/dedup/SparkUpdateEntity.java

@@ -69,6 +69,10 @@ public class SparkUpdateEntity extends AbstractSparkAction {
         final String workingPath = parser.get("workingPath");
         final String dedupGraphPath = parser.get("dedupGraphPath");
 
+        System.out.println(String.format("graphBasePath:  '%s'", graphBasePath));
+        System.out.println(String.format("workingPath:    '%s'", workingPath));
+        System.out.println(String.format("dedupGraphPath: '%s'", dedupGraphPath));
+
         final JavaSparkContext sc = new JavaSparkContext(spark.sparkContext());
 
         //for each entity