[Enrichment Step] get rid of hive

dhp-workflows/dhp-enrichment/src/main/java/eu/dnetlib/dhp/orcidtoresultfromsemrel/PrepareResultOrcidAssociationStep0.java

@@ -0,0 +1,90 @@
+
+package eu.dnetlib.dhp.orcidtoresultfromsemrel;
+
+import static eu.dnetlib.dhp.PropagationConstant.*;
+import static eu.dnetlib.dhp.PropagationConstant.readPath;
+import static eu.dnetlib.dhp.common.SparkSessionSupport.runWithSparkSession;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.spark.SparkConf;
+import org.apache.spark.api.java.function.FilterFunction;
+import org.apache.spark.api.java.function.MapFunction;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Encoders;
+import org.apache.spark.sql.SaveMode;
+import org.apache.spark.sql.SparkSession;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.gson.Gson;
+
+import eu.dnetlib.dhp.application.ArgumentApplicationParser;
+import eu.dnetlib.dhp.schema.oaf.Author;
+import eu.dnetlib.dhp.schema.oaf.Relation;
+import eu.dnetlib.dhp.schema.oaf.Result;
+import eu.dnetlib.dhp.schema.oaf.StructuredProperty;
+import scala.Tuple2;
+
+public class PrepareResultOrcidAssociationStep0 implements Serializable {
+
+	private static final Logger log = LoggerFactory.getLogger(PrepareResultOrcidAssociationStep0.class);
+
+	public static void main(String[] args) throws Exception {
+		String jsonConf = IOUtils
+			.toString(
+				PrepareResultOrcidAssociationStep0.class
+					.getResourceAsStream(
+						"/eu/dnetlib/dhp/orcidtoresultfromsemrel/input_prepareorcidtoresult0_parameters.json"));
+
+		final ArgumentApplicationParser parser = new ArgumentApplicationParser(jsonConf);
+		parser.parseArgument(args);
+
+		Boolean isSparkSessionManaged = isSparkSessionManaged(parser);
+		log.info("isSparkSessionManaged: {}", isSparkSessionManaged);
+
+		String inputPath = parser.get("sourcePath");
+		log.info("inputPath: {}", inputPath);
+
+		final String outputPath = parser.get("outputPath");
+		log.info("outputPath: {}", outputPath);
+
+		final List<String> allowedsemrel = Arrays
+			.stream(parser.get("allowedsemrels").split(";"))
+			.map(s -> s.toLowerCase())
+			.collect(Collectors.toList());
+
+		log.info("allowedSemRel: {}", new Gson().toJson(allowedsemrel));
+
+		SparkConf conf = new SparkConf();
+
+		runWithSparkSession(
+			conf,
+			isSparkSessionManaged,
+			spark -> {
+
+				selectRelations(
+					spark, inputPath, outputPath, allowedsemrel);
+			});
+	}
+
+	private static void selectRelations(SparkSession spark, String inputPath, String outputPath,
+		List<String> allowedsemrel) {
+
+		readPath(spark, inputPath, Relation.class)
+			.filter(
+				(FilterFunction<Relation>) r -> !r.getDataInfo().getDeletedbyinference()
+					&& allowedsemrel.contains(r.getRelClass().toLowerCase()))
+			.write()
+			.mode(SaveMode.Overwrite)
+			.option("compression", "gzip")
+			.json(outputPath);
+	}
+
+}

dhp-workflows/dhp-enrichment/src/main/java/eu/dnetlib/dhp/orcidtoresultfromsemrel/PrepareResultOrcidAssociationStep1.java

@@ -53,7 +53,7 @@ public class PrepareResultOrcidAssociationStep1 {
 		String inputPath = parser.get("sourcePath");
 		log.info("inputPath: {}", inputPath);
 
-		final String outputPath = parser.get("outputPath");
+		final String outputPath = parser.get("workingPath");
 		log.info("outputPath: {}", outputPath);
 
 		final String resultClassName = parser.get("resultTableName");
@@ -76,15 +76,6 @@ public class PrepareResultOrcidAssociationStep1 {
 
 		SparkConf conf = new SparkConf();
 
-		String inputRelationPath = inputPath + "/relation";
-		log.info("inputRelationPath: {}", inputRelationPath);
-
-		String inputResultPath = inputPath + "/" + resultType;
-		log.info("inputResultPath: {}", inputResultPath);
-
-		String outputResultPath = outputPath + "/" + resultType;
-		log.info("outputResultPath: {}", outputResultPath);
-
 		runWithSparkSession(
 			conf,
 			isSparkSessionManaged,
@@ -112,12 +103,14 @@ public class PrepareResultOrcidAssociationStep1 {
 			.write()
 			.mode(SaveMode.Overwrite)
 			.option("compression", "gzip")
-			.json(outputPath + "/relationSubset");
+			.json(outputPath + "/" + resultType + "/relationSubset");
 
 		Dataset<Relation> relation = readPath(spark, outputPath + "/relationSubset", Relation.class);
 
 		log.info("Reading Graph table from: {}", inputResultPath);
 
+		final String resultOutputPath = outputPath + "/resultSubset/" + resultType;
+
 		readPath(spark, inputResultPath, resultClazz)
 			.filter(
 				(FilterFunction<R>) r -> !r.getDataInfo().getDeletedbyinference() && !r.getDataInfo().getInvisible())
@@ -135,11 +128,11 @@ public class PrepareResultOrcidAssociationStep1 {
 			.write()
 			.mode(SaveMode.Overwrite)
 			.option("compression", "gzip")
-			.json(outputPath + "/resultSubset");
+			.json(resultOutputPath);
 
-		Dataset<R> result = readPath(spark, outputPath + "/resultSubset", resultClazz);
+		Dataset<R> result = readPath(spark, resultOutputPath, resultClazz);
 
-		result.foreach((ForeachFunction<R>) r -> System.out.println(new ObjectMapper().writeValueAsString(r)));
+		// result.foreach((ForeachFunction<R>) r -> System.out.println(new ObjectMapper().writeValueAsString(r)));
 
 		result
 			.joinWith(relation, result.col("id").equalTo(relation.col("source")))

dhp-workflows/dhp-enrichment/src/main/java/eu/dnetlib/dhp/orcidtoresultfromsemrel/PrepareResultOrcidAssociationStep2.java

@@ -52,7 +52,7 @@ public class PrepareResultOrcidAssociationStep2 {
 			conf,
 			isSparkSessionManaged,
 			spark -> {
-				removeOutputDir(spark, outputPath);
+				// removeOutputDir(spark, outputPath);
 				mergeInfo(spark, inputPath, outputPath);
 			});
 	}

dhp-workflows/dhp-enrichment/src/main/resources/eu/dnetlib/dhp/orcidtoresultfromsemrel/input_prepareorcidtoresult0_parameters.json

@@ -0,0 +1,26 @@
+[
+  {
+    "paramName":"s",
+    "paramLongName":"sourcePath",
+    "paramDescription": "the path of the sequencial file to read",
+    "paramRequired": true
+  },
+  {
+    "paramName":"as",
+    "paramLongName":"allowedsemrels",
+    "paramDescription": "the allowed sematinc relations for propagation",
+    "paramRequired": true
+  },
+  {
+    "paramName": "out",
+    "paramLongName": "outputPath",
+    "paramDescription": "the path used to store temporary output files",
+    "paramRequired": true
+  },
+  {
+    "paramName": "ssm",
+    "paramLongName": "isSparkSessionManaged",
+    "paramDescription": "true if the spark session is managed, false otherwise",
+    "paramRequired": false
+  }
+]

dhp-workflows/dhp-enrichment/src/main/resources/eu/dnetlib/dhp/orcidtoresultfromsemrel/input_prepareorcidtoresult_parameters.json

@@ -18,8 +18,8 @@
     "paramRequired": true
   },
   {
-    "paramName": "out",
+    "paramName": "wp",
-    "paramLongName": "outputPath",
+    "paramLongName": "workingPath",
     "paramDescription": "the path used to store temporary output files",
     "paramRequired": true
   },

dhp-workflows/dhp-enrichment/src/main/resources/eu/dnetlib/dhp/wf/subworkflows/orcidtoresultfromsemrel/oozie_app/workflow.xml

@@ -14,7 +14,7 @@
         </property>
     </parameters>
 
-    <start to="reset_outputpath"/>
+    <start to="fork_prepare_assoc_step1"/>
 
     <kill name="Kill">
         <message>Action failed, error message[${wf:errorMessage(wf:lastErrorNode())}]</message>
@@ -80,7 +80,37 @@
         <error to="Kill"/>
     </action>
 
-    <join name="copy_wait" to="fork_prepare_assoc_step1"/>
+    <join name="copy_wait" to="prepare_relations"/>
+
+    <action name="prepare_relations">
+        <spark xmlns="uri:oozie:spark-action:0.2">
+            <master>yarn</master>
+            <mode>cluster</mode>
+            <name>ORCIDPropagation-PreparePhase0-SelectRELATIONS</name>
+            <class>eu.dnetlib.dhp.orcidtoresultfromsemrel.PrepareResultOrcidAssociationStep0</class>
+            <jar>dhp-enrichment-${projectVersion}.jar</jar>
+            <spark-opts>
+                --executor-cores=${sparkExecutorCores}
+                --executor-memory=${sparkExecutorMemory}
+                --driver-memory=${sparkDriverMemory}
+                --conf spark.extraListeners=${spark2ExtraListeners}
+                --conf spark.sql.queryExecutionListeners=${spark2SqlQueryExecutionListeners}
+                --conf spark.yarn.historyServer.address=${spark2YarnHistoryServerAddress}
+                --conf spark.eventLog.dir=${nameNode}${spark2EventLogDir}
+                --conf spark.dynamicAllocation.enabled=true
+                --conf spark.dynamicAllocation.maxExecutors=${spark2MaxExecutors}
+                --conf spark.sql.shuffle.partitions=3840
+                --conf spark.speculation=false
+                --conf spark.hadoop.mapreduce.map.speculative=false
+                --conf spark.hadoop.mapreduce.reduce.speculative=false
+            </spark-opts>
+            <arg>--sourcePath</arg><arg>${sourcePath}/relation</arg>
+            <arg>--outputPath</arg><arg>${workingDir}/orcidprop/relationSubset</arg>
+            <arg>--allowedsemrels</arg><arg>${allowedsemrels}</arg>
+        </spark>
+        <ok to="wait"/>
+        <error to="Kill"/>
+    </action>
 
     <fork name="fork_prepare_assoc_step1">
         <path start="join_prepare_publication"/>
@@ -113,7 +143,7 @@
             </spark-opts>
             <arg>--sourcePath</arg><arg>${sourcePath}</arg>
             <arg>--resultTableName</arg><arg>eu.dnetlib.dhp.schema.oaf.Publication</arg>
-            <arg>--outputPath</arg><arg>${workingDir}/orcidprop/preparedInfo/targetOrcidAssoc</arg>
+            <arg>--workingPath</arg><arg>${workingDir}/orcidprop</arg>
             <arg>--allowedsemrels</arg><arg>${allowedsemrels}</arg>
             <arg>--allowedpids</arg><arg>${allowedpids}</arg>
         </spark>
@@ -141,7 +171,7 @@
             </spark-opts>
             <arg>--sourcePath</arg><arg>${sourcePath}</arg>
             <arg>--resultTableName</arg><arg>eu.dnetlib.dhp.schema.oaf.Dataset</arg>
-            <arg>--outputPath</arg><arg>${workingDir}/orcidprop/preparedInfo/targetOrcidAssoc</arg>
+            <arg>--workingPath</arg><arg>${workingDir}/orcidprop</arg>
             <arg>--allowedsemrels</arg><arg>${allowedsemrels}</arg>
             <arg>--allowedpids</arg><arg>${allowedpids}</arg>
         </spark>
@@ -169,7 +199,7 @@
             </spark-opts>
             <arg>--sourcePath</arg><arg>${sourcePath}</arg>
             <arg>--resultTableName</arg><arg>eu.dnetlib.dhp.schema.oaf.OtherResearchProduct</arg>
-            <arg>--outputPath</arg><arg>${workingDir}/orcidprop/preparedInfo/targetOrcidAssoc</arg>
+            <arg>--workingPath</arg><arg>${workingDir}/orcidprop</arg>
             <arg>--allowedsemrels</arg><arg>${allowedsemrels}</arg>
             <arg>--allowedpids</arg><arg>${allowedpids}</arg>
         </spark>
@@ -197,7 +227,7 @@
             </spark-opts>
             <arg>--sourcePath</arg><arg>${sourcePath}</arg>
             <arg>--resultTableName</arg><arg>eu.dnetlib.dhp.schema.oaf.Software</arg>
-            <arg>--outputPath</arg><arg>${workingDir}/orcidprop/preparedInfo/targetOrcidAssoc</arg>
+            <arg>--workingPath</arg><arg>${workingDir}/orcidprop</arg>
             <arg>--allowedsemrels</arg><arg>${allowedsemrels}</arg>
             <arg>--allowedpids</arg><arg>${allowedpids}</arg>
         </spark>
@@ -225,8 +255,8 @@
                 --conf spark.dynamicAllocation.enabled=true
                 --conf spark.dynamicAllocation.maxExecutors=${spark2MaxExecutors}
             </spark-opts>
-            <arg>--sourcePath</arg><arg>${workingDir}/orcidprop/preparedInfo/targetOrcidAssoc</arg>
+            <arg>--sourcePath</arg><arg>${workingDir}/orcidprop</arg>
-            <arg>--outputPath</arg><arg>${workingDir}/orcidprop/preparedInfo/mergedOrcidAssoc</arg>
+            <arg>--outputPath</arg><arg>${workingDir}/orcidprop//mergedOrcidAssoc</arg>
         </spark>
         <ok to="fork-join-exec-propagation"/>
         <error to="Kill"/>
@@ -261,7 +291,7 @@
                 --conf spark.hadoop.mapreduce.reduce.speculative=false
                 --conf spark.sql.shuffle.partitions=3840
             </spark-opts>
-            <arg>--possibleUpdatesPath</arg><arg>${workingDir}/orcidprop/preparedInfo/mergedOrcidAssoc</arg>
+            <arg>--possibleUpdatesPath</arg><arg>${workingDir}/orcidprop/mergedOrcidAssoc</arg>
             <arg>--sourcePath</arg><arg>${sourcePath}/publication</arg>
             <arg>--resultTableName</arg><arg>eu.dnetlib.dhp.schema.oaf.Publication</arg>
             <arg>--outputPath</arg><arg>${outputPath}/publication</arg>
@@ -291,7 +321,7 @@
                 --conf spark.hadoop.mapreduce.map.speculative=false
                 --conf spark.hadoop.mapreduce.reduce.speculative=false
             </spark-opts>
-            <arg>--possibleUpdatesPath</arg><arg>${workingDir}/orcidprop/preparedInfo/mergedOrcidAssoc</arg>
+            <arg>--possibleUpdatesPath</arg><arg>${workingDir}/orcidprop/mergedOrcidAssoc</arg>
             <arg>--sourcePath</arg><arg>${sourcePath}/dataset</arg>
             <arg>--resultTableName</arg><arg>eu.dnetlib.dhp.schema.oaf.Dataset</arg>
             <arg>--outputPath</arg><arg>${outputPath}/dataset</arg>
@@ -321,7 +351,7 @@
                 --conf spark.hadoop.mapreduce.map.speculative=false
                 --conf spark.hadoop.mapreduce.reduce.speculative=false
             </spark-opts>
-            <arg>--possibleUpdatesPath</arg><arg>${workingDir}/orcidprop/preparedInfo/mergedOrcidAssoc</arg>
+            <arg>--possibleUpdatesPath</arg><arg>${workingDir}/orcidprop/mergedOrcidAssoc</arg>
             <arg>--sourcePath</arg><arg>${sourcePath}/otherresearchproduct</arg>
             <arg>--resultTableName</arg><arg>eu.dnetlib.dhp.schema.oaf.OtherResearchProduct</arg>
             <arg>--outputPath</arg><arg>${outputPath}/otherresearchproduct</arg>
@@ -351,7 +381,7 @@
                 --conf spark.hadoop.mapreduce.map.speculative=false
                 --conf spark.hadoop.mapreduce.reduce.speculative=false
             </spark-opts>
-            <arg>--possibleUpdatesPath</arg><arg>${workingDir}/orcidprop/preparedInfo/mergedOrcidAssoc</arg>
+            <arg>--possibleUpdatesPath</arg><arg>${workingDir}/orcidprop/mergedOrcidAssoc</arg>
             <arg>--sourcePath</arg><arg>${sourcePath}/software</arg>
             <arg>--resultTableName</arg><arg>eu.dnetlib.dhp.schema.oaf.Software</arg>
             <arg>--outputPath</arg><arg>${outputPath}/software</arg>