new workflow and parameters for country propagation

dhp-workflows/dhp-propagation/src/main/resources/eu/dnetlib/dhp/countrypropagation/input_countrypropagation_parameters.json

@@ -11,18 +11,6 @@
     "paramDescription": "the path of the sequencial file to read",
     "paramRequired": true
   },
-  {
-    "paramName":"wl",
-    "paramLongName":"whitelist",
-    "paramDescription": "datasource id that will be considered even if not in the allowed typology list. Split by ;",
-    "paramRequired": true
-  },
-  {
-    "paramName":"at",
-    "paramLongName":"allowedtypes",
-    "paramDescription": "the types of the allowed datasources. Split by ;",
-    "paramRequired": true
-  },
   {
     "paramName":"h",
     "paramLongName":"hive_metastore_uris",
@@ -40,5 +28,11 @@
     "paramLongName":"saveGraph",
     "paramDescription": "true if the new version of the graph must be saved",
     "paramRequired": true
+  },
+  {
+    "paramName":"tn",
+    "paramLongName":"resultTableName",
+    "paramDescription": "the name of the result table we are currently working on",
+    "paramRequired": true
   }
 ]

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

@@ -38,12 +38,152 @@
         </property>
     </parameters>
 
-    <start to="CountryPropagation"/>
+    <start to="prepare_datasource_country_association"/>
 
     <kill name="Kill">
         <message>Action failed, error message[${wf:errorMessage(wf:lastErrorNode())}]</message>
     </kill>
 
+    <action name="prepare_datasource_country_association">
+        <spark xmlns="uri:oozie:spark-action:0.2">
+            <master>yarn</master>
+            <mode>cluster</mode>
+            <name>PrepareDatasourceCountryAssociation</name>
+            <class>eu.dnetlib.dhp.countrypropagation.PrepareDatasourceCountryAssociation</class>
+            <jar>dhp-graph-propagation-${projectVersion}.jar</jar>
+            <spark-opts>
+                --executor-cores=${sparkExecutorCoresForJoining}
+                --executor-memory=${sparkExecutorMemoryForJoining}
+                --driver-memory=${sparkDriverMemoryForJoining}
+                --conf spark.extraListeners=${spark2ExtraListeners}
+                --conf spark.sql.queryExecutionListeners=${spark2SqlQueryExecutionListeners}
+                --conf spark.yarn.historyServer.address=${spark2YarnHistoryServerAddress}
+                --conf spark.eventLog.dir=${nameNode}${spark2EventLogDir}
+            </spark-opts>
+            <arg>-mt</arg> <arg>yarn-cluster</arg>
+            <arg>--sourcePath</arg><arg>${sourcePath}</arg>
+            <arg>--whitelist</arg><arg>${whitelist}</arg>
+            <arg>--allowedtypes</arg><arg>${allowedtypes}</arg>
+            <arg>--hive_metastore_uris</arg><arg>${hive_metastore_uris}</arg>
+        </spark>
+        <ok to="fork_join_apply_country_propagation"/>
+        <error to="Kill"/>
+    </action>
+    <fork name="fork_join_apply_country_propagation">
+        <path start="join_propagation_publication"/>
+        <path start="join_propagation_dataset"/>
+        <path start="join_propagation_otherresearchproduct"/>
+        <path start="join_propagation_software"/>
+    </fork>
+
+    <action name="join_propagation_publication">
+        <spark xmlns="uri:oozie:spark-action:0.2">
+            <master>yarn</master>
+            <mode>cluster</mode>
+            <name>countryPropagationForPublications</name>
+            <class>eu.dnetlib.dhp.countrypropagation.SparkCountryPropagationJob2</class>
+            <jar>dhp-graph-propagation-${projectVersion}.jar</jar>
+            <spark-opts>
+                --executor-cores=${sparkExecutorCoresForJoining}
+                --executor-memory=${sparkExecutorMemoryForJoining}
+                --driver-memory=${sparkDriverMemoryForJoining}
+                --conf spark.extraListeners=${spark2ExtraListeners}
+                --conf spark.sql.queryExecutionListeners=${spark2SqlQueryExecutionListeners}
+                --conf spark.yarn.historyServer.address=${spark2YarnHistoryServerAddress}
+                --conf spark.eventLog.dir=${nameNode}${spark2EventLogDir}
+            </spark-opts>
+            <arg>-mt</arg> <arg>yarn-cluster</arg>
+            <arg>--sourcePath</arg><arg>${sourcePath}</arg>
+            <arg>--hive_metastore_uris</arg><arg>${hive_metastore_uris}</arg>
+            <arg>--writeUpdate</arg><arg>${writeUpdate}</arg>
+            <arg>--saveGraph</arg><arg>${saveGraph}</arg>
+            <arg>--resultTableName</arg><arg>eu.dnetlib.dhp.schema.oaf.Publication</arg>
+        </spark>
+        <ok to="end"/>
+        <error to="Kill"/>
+    </action>
+
+    <action name="join_propagation_dataset">
+        <spark xmlns="uri:oozie:spark-action:0.2">
+            <master>yarn</master>
+            <mode>cluster</mode>
+            <name>countryPropagationForDataset</name>
+            <class>eu.dnetlib.dhp.countrypropagation.SparkCountryPropagationJob2</class>
+            <jar>dhp-graph-propagation-${projectVersion}.jar</jar>
+            <spark-opts>
+                --executor-cores=${sparkExecutorCoresForJoining}
+                --executor-memory=${sparkExecutorMemoryForJoining}
+                --driver-memory=${sparkDriverMemoryForJoining}
+                --conf spark.extraListeners=${spark2ExtraListeners}
+                --conf spark.sql.queryExecutionListeners=${spark2SqlQueryExecutionListeners}
+                --conf spark.yarn.historyServer.address=${spark2YarnHistoryServerAddress}
+                --conf spark.eventLog.dir=${nameNode}${spark2EventLogDir}
+            </spark-opts>
+            <arg>-mt</arg> <arg>yarn-cluster</arg>
+            <arg>--sourcePath</arg><arg>${sourcePath}</arg>
+            <arg>--hive_metastore_uris</arg><arg>${hive_metastore_uris}</arg>
+            <arg>--writeUpdate</arg><arg>${writeUpdate}</arg>
+            <arg>--saveGraph</arg><arg>${saveGraph}</arg>
+            <arg>--resultTableName</arg><arg>eu.dnetlib.dhp.schema.oaf.Dataset</arg>
+        </spark>
+        <ok to="end"/>
+        <error to="Kill"/>
+    </action>
+
+    <action name="join_propagation_otherresearchproduct">
+        <spark xmlns="uri:oozie:spark-action:0.2">
+            <master>yarn</master>
+            <mode>cluster</mode>
+            <name>countryPropagationForORP</name>
+            <class>eu.dnetlib.dhp.countrypropagation.SparkCountryPropagationJob2</class>
+            <jar>dhp-graph-propagation-${projectVersion}.jar</jar>
+            <spark-opts>
+                --executor-cores=${sparkExecutorCoresForJoining}
+                --executor-memory=${sparkExecutorMemoryForJoining}
+                --driver-memory=${sparkDriverMemoryForJoining}
+                --conf spark.extraListeners=${spark2ExtraListeners}
+                --conf spark.sql.queryExecutionListeners=${spark2SqlQueryExecutionListeners}
+                --conf spark.yarn.historyServer.address=${spark2YarnHistoryServerAddress}
+                --conf spark.eventLog.dir=${nameNode}${spark2EventLogDir}
+            </spark-opts>
+            <arg>-mt</arg> <arg>yarn-cluster</arg>
+            <arg>--sourcePath</arg><arg>${sourcePath}</arg>
+            <arg>--hive_metastore_uris</arg><arg>${hive_metastore_uris}</arg>
+            <arg>--writeUpdate</arg><arg>${writeUpdate}</arg>
+            <arg>--saveGraph</arg><arg>${saveGraph}</arg>
+            <arg>--resultTableName</arg><arg>eu.dnetlib.dhp.schema.oaf.Otherresearchproduct</arg>
+        </spark>
+        <ok to="end"/>
+        <error to="Kill"/>
+    </action>
+
+    <action name="join_propagation_software">
+        <spark xmlns="uri:oozie:spark-action:0.2">
+            <master>yarn</master>
+            <mode>cluster</mode>
+            <name>countryPropagationForSoftware</name>
+            <class>eu.dnetlib.dhp.countrypropagation.SparkCountryPropagationJob2</class>
+            <jar>dhp-graph-propagation-${projectVersion}.jar</jar>
+            <spark-opts>
+                --executor-cores=${sparkExecutorCoresForJoining}
+                --executor-memory=${sparkExecutorMemoryForJoining}
+                --driver-memory=${sparkDriverMemoryForJoining}
+                --conf spark.extraListeners=${spark2ExtraListeners}
+                --conf spark.sql.queryExecutionListeners=${spark2SqlQueryExecutionListeners}
+                --conf spark.yarn.historyServer.address=${spark2YarnHistoryServerAddress}
+                --conf spark.eventLog.dir=${nameNode}${spark2EventLogDir}
+            </spark-opts>
+            <arg>-mt</arg> <arg>yarn-cluster</arg>
+            <arg>--sourcePath</arg><arg>${sourcePath}</arg>
+            <arg>--hive_metastore_uris</arg><arg>${hive_metastore_uris}</arg>
+            <arg>--writeUpdate</arg><arg>${writeUpdate}</arg>
+            <arg>--saveGraph</arg><arg>${saveGraph}</arg>
+            <arg>--resultTableName</arg><arg>eu.dnetlib.dhp.schema.oaf.Software</arg>
+        </spark>
+        <ok to="end"/>
+        <error to="Kill"/>
+    </action>
+    
     <action name="CountryPropagation">
         <spark xmlns="uri:oozie:spark-action:0.2">
             <job-tracker>${jobTracker}</job-tracker>