minor

dhp-workflows/dhp-blacklist/pom.xml

@@ -11,6 +11,7 @@
 
     <artifactId>dhp-blacklist</artifactId>
     <dependencies>
+
         <dependency>
             <groupId>eu.dnetlib.dhp</groupId>
             <artifactId>dhp-graph-mapper</artifactId>

dhp-workflows/dhp-blacklist/src/main/java/eu/dnetlib/dhp/blacklist/PrepareMergedRelationJob.java

@@ -1,9 +1,10 @@
+
 package eu.dnetlib.dhp.blacklist;
 
-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 static eu.dnetlib.dhp.common.SparkSessionSupport.runWithSparkHiveSession;
+
+import java.util.Optional;
+
 import org.apache.commons.io.IOUtils;
 import org.apache.spark.SparkConf;
 import org.apache.spark.api.java.function.MapFunction;
@@ -14,11 +15,10 @@ import org.apache.spark.sql.SparkSession;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.util.Arrays;
-import java.util.List;
-import java.util.Optional;
+import com.fasterxml.jackson.databind.ObjectMapper;
 
-import static eu.dnetlib.dhp.common.SparkSessionSupport.runWithSparkHiveSession;
+import eu.dnetlib.dhp.application.ArgumentApplicationParser;
+import eu.dnetlib.dhp.schema.oaf.Relation;
 
 public class PrepareMergedRelationJob {
 
@@ -47,7 +47,7 @@ public class PrepareMergedRelationJob {
 		log.info("inputPath: {}", inputPath);
 
 		String outputPath = parser.get("outputPath");
-        log.info("outputPath: {} " , outputPath);
+		log.info("outputPath: {} ", outputPath);
 
 		SparkConf conf = new SparkConf();
 		conf.set("hive.metastore.uris", parser.get("hive_metastore_uris"));
@@ -56,8 +56,6 @@ public class PrepareMergedRelationJob {
 			conf,
 			isSparkSessionManaged,
 			spark -> {
-                    // removeOutputDir(spark, potentialUpdatePath);
-                    // removeOutputDir(spark, alreadyLinkedPath);
 				selectMergesRelations(
 					spark,
 					inputPath,
@@ -70,12 +68,15 @@ public class PrepareMergedRelationJob {
 		Dataset<Relation> relation = readRelations(spark, inputPath);
 		relation.createOrReplaceTempView("relation");
 
-        spark.sql("Select * from relation where relclass = 'merges' and datainfo.deletedbyinference = false")
+		spark
+			.sql("Select * from relation " +
+					"where relclass = 'merges' " +
+					"and datainfo.deletedbyinference = false")
 			.as(Encoders.bean(Relation.class))
 			.toJSON()
 			.write()
 			.mode(SaveMode.Overwrite)
-                .option("compression","gzip")
+			.option("compression", "gzip")
 			.text(outputPath);
 	}
 

dhp-workflows/dhp-blacklist/src/main/java/eu/dnetlib/dhp/blacklist/SparkRemoveBlacklistedRelationJob.java

@@ -55,7 +55,6 @@ public class SparkRemoveBlacklistedRelationJob {
 		log.info("mergesPath {}: ", mergesPath);
 
 		SparkConf conf = new SparkConf();
-		conf.set("hive.metastore.uris", parser.get("hive_metastore_uris"));
 
 		runWithSparkSession(
 				conf,
@@ -68,6 +67,8 @@ public class SparkRemoveBlacklistedRelationJob {
 							outputPath,
 							mergesPath);
 				});
+
+
 	}
 
 	private static void removeBlacklistedRelations(SparkSession spark, String blacklistPath, String inputPath,
@@ -78,7 +79,6 @@ public class SparkRemoveBlacklistedRelationJob {
 
 		Dataset<Relation> dedupSource = blackListed
 			.joinWith(mergesRelation, blackListed.col("source").equalTo(mergesRelation.col("target")), "left_outer")
-			// .joinWith(inputRelation,blackListed.col("target").equalTo(inputRelation.col("target")),"left_outer")
 			.map(c -> {
 				Optional<Relation> merged = Optional.ofNullable(c._2());
 				Relation bl = c._1();

dhp-workflows/dhp-blacklist/src/main/resources/eu/dnetlib/dhp/blacklist/oozie_app/workflow.xml

@@ -63,7 +63,7 @@
                 --conf spark.eventLog.dir=${nameNode}${spark2EventLogDir}
             </spark-opts>
             <arg>--sourcePath</arg><arg>${sourcePath}</arg>
-            <arg>--outputPath</arg><arg>${workingDir}/relation</arg>
+            <arg>--outputPath</arg><arg>${workingDir}/mergesRelation</arg>
             <arg>--hive_metastore_uris</arg><arg>${hive_metastore_uris}</arg>
         </spark>
         <ok to="apply_blacklist"/>
@@ -89,6 +89,7 @@
             <arg>--sourcePath</arg><arg>${sourcePath}</arg>
             <arg>--outputPath</arg><arg>${workingDir}/relation</arg>
             <arg>--hdfsPath</arg><arg>${workingDir}/blacklist</arg>
+            <arg>--mergesPath</arg><arg>${workingDir}/mergesRelation</arg>
         </spark>
         <ok to="End"/>
         <error to="Kill"/>

dhp-workflows/dhp-blacklist/src/main/resources/eu/dnetlib/dhp/blacklist/sparkblacklist_parameters.json

@@ -22,5 +22,12 @@
 		"paramLongName": "isSparkSessionManaged",
 		"paramDescription": "true if the spark session is managed",
 		"paramRequired": false
+	},
+	{
+		"paramName": "m",
+		"paramLongName": "mergesPath",
+		"paramDescription": "true if the spark session is managed",
+		"paramRequired": true
+
 	}
 ]