updated workflow for generation of Scholix Datasource's to use mdstore transactions

dhp-workflows/dhp-aggregation/src/main/resources/eu/dnetlib/dhp/sx/bio/db/oozie_app/workflow.xml

@@ -1,4 +1,4 @@
-    <workflow-app name="Transform_BioEntity_Workflow" xmlns="uri:oozie:workflow:0.5">
+<workflow-app name="Transform_BioEntity_Workflow" xmlns="uri:oozie:workflow:0.5">
     <parameters>
         <property>
             <name>sourcePath</name>
@@ -8,19 +8,40 @@
             <name>database</name>
             <description>the PDB Database Working Path</description>
         </property>
-
         <property>
-            <name>targetPath</name>
+            <name>mdStoreOutputId</name>
-            <description>the Target Working dir path</description>
+            <description>the identifier of the cleaned MDStore</description>
+        </property>
+        <property>
+            <name>mdStoreManagerURI</name>
+            <description>the path of the cleaned mdstore</description>
         </property>
     </parameters>
 
-    <start to="ConvertDB"/>
+    <start to="StartTransaction"/>
+
 
     <kill name="Kill">
         <message>Action failed, error message[${wf:errorMessage(wf:lastErrorNode())}]</message>
     </kill>
 
+    <action name="StartTransaction">
+        <java>
+            <configuration>
+                <property>
+                    <name>oozie.launcher.mapreduce.user.classpath.first</name>
+                    <value>true</value>
+                </property>
+            </configuration>
+            <main-class>eu.dnetlib.dhp.aggregation.mdstore.MDStoreActionNode</main-class>
+            <arg>--action</arg><arg>NEW_VERSION</arg>
+            <arg>--mdStoreID</arg><arg>${mdStoreOutputId}</arg>
+            <arg>--mdStoreManagerURI</arg><arg>${mdStoreManagerURI}</arg>
+            <capture-output/>
+        </java>
+        <ok to="ConvertDB"/>
+        <error to="RollBack"/>
+    </action>
     <action name="ConvertDB">
         <spark xmlns="uri:oozie:spark-action:0.2">
             <master>yarn</master>
@@ -41,11 +62,48 @@
             <arg>--master</arg><arg>yarn</arg>
             <arg>--dbPath</arg><arg>${sourcePath}</arg>
             <arg>--database</arg><arg>${database}</arg>
-            <arg>--targetPath</arg><arg>${targetPath}</arg>
+            <arg>--mdstoreOutputVersion</arg><arg>${wf:actionData('StartTransaction')['mdStoreVersion']}</arg>
         </spark>
-        <ok to="End"/>
+        <ok to="CommitVersion"/>
-        <error to="Kill"/>
+        <error to="RollBack"/>
+
     </action>
-    <end name="End"/>
+        <action name="CommitVersion">
+            <java>
+                <configuration>
+                    <property>
+                        <name>oozie.launcher.mapreduce.user.classpath.first</name>
+                        <value>true</value>
+                    </property>
+                </configuration>
+                <main-class>eu.dnetlib.dhp.aggregation.mdstore.MDStoreActionNode</main-class>
+                <arg>--action</arg><arg>COMMIT</arg>
+                <arg>--namenode</arg><arg>${nameNode}</arg>
+                <arg>--mdStoreVersion</arg><arg>${wf:actionData('StartTransaction')['mdStoreVersion']}</arg>
+                <arg>--mdStoreManagerURI</arg><arg>${mdStoreManagerURI}</arg>
+            </java>
+            <ok to="End"/>
+            <error to="Kill"/>
+        </action>
+
+        <action name="RollBack">
+            <java>
+                <configuration>
+                    <property>
+                        <name>oozie.launcher.mapreduce.user.classpath.first</name>
+                        <value>true</value>
+                    </property>
+                </configuration>
+                <main-class>eu.dnetlib.dhp.aggregation.mdstore.MDStoreActionNode</main-class>
+                <arg>--action</arg><arg>ROLLBACK</arg>
+                <arg>--mdStoreVersion</arg><arg>${wf:actionData('StartTransaction')['mdStoreVersion']}</arg>
+                <arg>--mdStoreManagerURI</arg><arg>${mdStoreManagerURI}</arg>
+            </java>
+            <ok to="Kill"/>
+            <error to="Kill"/>
+        </action>
+
+
+        <end name="End"/>
 
 </workflow-app>

dhp-workflows/dhp-aggregation/src/main/resources/eu/dnetlib/dhp/sx/bio/ebi/bio_to_oaf_params.json

@@ -2,5 +2,5 @@
   {"paramName":"mt",  "paramLongName":"master",       "paramDescription": "should be local or yarn",                  "paramRequired": true},
   {"paramName":"db",  "paramLongName":"database",     "paramDescription": "should be PDB or UNIPROT",                 "paramRequired": true},
   {"paramName":"p",   "paramLongName":"dbPath",       "paramDescription": "the path of the database to transform",    "paramRequired": true},
-  {"paramName":"t",   "paramLongName":"targetPath",   "paramDescription": "the OAF target path ",                     "paramRequired": true}
+  {"paramName":"mo",   "paramLongName":"mdstoreOutputVersion",     "paramDescription": "the oaf path ",                "paramRequired": true}
 ]

dhp-workflows/dhp-aggregation/src/main/resources/eu/dnetlib/dhp/sx/bio/ebi/ebi_to_df_params.json

@@ -1,5 +1,20 @@
 [
-  {"paramName":"mt",  "paramLongName":"master",     "paramDescription": "should be local or yarn",                  "paramRequired": true},
+  {
-  {"paramName":"s",   "paramLongName":"sourcePath","paramDescription": "the source Path",                              "paramRequired": true},
+    "paramName": "mt",
-  {"paramName":"t",   "paramLongName":"targetPath","paramDescription": "the  oaf path ",  "paramRequired": true}
+    "paramLongName": "master",
+    "paramDescription": "should be local or yarn",
+    "paramRequired": true
+  },
+  {
+    "paramName": "s",
+    "paramLongName": "sourcePath",
+    "paramDescription": "the source Path",
+    "paramRequired": true
+  },
+  {
+    "paramName": "mo",
+    "paramLongName": "mdstoreOutputVersion",
+    "paramDescription": "the oaf path ",
+    "paramRequired": true
+  }
 ]

dhp-workflows/dhp-aggregation/src/main/resources/eu/dnetlib/dhp/sx/bio/ebi/oozie_app/workflow.xml

@@ -9,34 +9,26 @@
             <description>the Working Path</description>
         </property>
         <property>
-            <name>targetPath</name>
+            <name>mdStoreOutputId</name>
-            <description>the OAF MDStore Path</description>
+            <description>the identifier of the cleaned MDStore</description>
         </property>
         <property>
-            <name>sparkDriverMemory</name>
+            <name>mdStoreManagerURI</name>
-            <description>memory for driver process</description>
+            <description>the path of the cleaned mdstore</description>
-        </property>
-        <property>
-            <name>sparkExecutorMemory</name>
-            <description>memory for individual executor</description>
-        </property>
-        <property>
-            <name>sparkExecutorCores</name>
-            <description>number of cores used by single executor</description>
         </property>
         <property>
             <name>resumeFrom</name>
-            <value>DownloadEBILinks</value>
+            <value>CreateEBIDataSet</value>
             <description>node to start</description>
         </property>
     </parameters>
 
-    <start to="resume_from"/>
+    <start to="StartTransaction"/>
 
     <decision name="resume_from">
         <switch>
             <case to="DownloadEBILinks">${wf:conf('resumeFrom') eq 'DownloadEBILinks'}</case>
-            <case to="CreateEBIDataSet">${wf:conf('resumeFrom') eq 'CreateEBIDataSet'}</case>
+            <case to="StartTransaction">${wf:conf('resumeFrom') eq 'CreateEBIDataSet'}</case>
             <default to="DownloadEBILinks"/>
         </switch>
     </decision>
@@ -77,9 +69,29 @@
             <move source="${sourcePath}/ebi_links_dataset" target="${sourcePath}/ebi_links_dataset_old"/>
             <move source="${workingPath}/links_final" target="${sourcePath}/ebi_links_dataset"/>
         </fs>
-        <ok to="CreateEBIDataSet"/>
+        <ok to="StartTransaction"/>
         <error to="Kill"/>
     </action>
+
+    <action name="StartTransaction">
+        <java>
+            <configuration>
+                <property>
+                    <name>oozie.launcher.mapreduce.user.classpath.first</name>
+                    <value>true</value>
+                </property>
+            </configuration>
+            <main-class>eu.dnetlib.dhp.aggregation.mdstore.MDStoreActionNode</main-class>
+            <arg>--action</arg><arg>NEW_VERSION</arg>
+            <arg>--mdStoreID</arg><arg>${mdStoreOutputId}</arg>
+            <arg>--mdStoreManagerURI</arg><arg>${mdStoreManagerURI}</arg>
+            <capture-output/>
+        </java>
+        <ok to="CreateEBIDataSet"/>
+        <error to="RollBack"/>
+    </action>
+
+
     <action name="CreateEBIDataSet">
         <spark xmlns="uri:oozie:spark-action:0.2">
             <master>yarn-cluster</master>
@@ -95,11 +107,49 @@
                 ${sparkExtraOPT}
             </spark-opts>
             <arg>--sourcePath</arg><arg>${sourcePath}/ebi_links_dataset</arg>
-            <arg>--targetPath</arg><arg>${targetPath}</arg>
+            <arg>--mdstoreOutputVersion</arg><arg>${wf:actionData('StartTransaction')['mdStoreVersion']}</arg>
             <arg>--master</arg><arg>yarn</arg>
         </spark>
         <ok to="End"/>
         <error to="Kill"/>
     </action>
+
+
+    <action name="CommitVersion">
+        <java>
+            <configuration>
+                <property>
+                    <name>oozie.launcher.mapreduce.user.classpath.first</name>
+                    <value>true</value>
+                </property>
+            </configuration>
+            <main-class>eu.dnetlib.dhp.aggregation.mdstore.MDStoreActionNode</main-class>
+            <arg>--action</arg><arg>COMMIT</arg>
+            <arg>--namenode</arg><arg>${nameNode}</arg>
+            <arg>--mdStoreVersion</arg><arg>${wf:actionData('StartTransaction')['mdStoreVersion']}</arg>
+            <arg>--mdStoreManagerURI</arg><arg>${mdStoreManagerURI}</arg>
+        </java>
+        <ok to="End"/>
+        <error to="Kill"/>
+    </action>
+
+    <action name="RollBack">
+        <java>
+            <configuration>
+                <property>
+                    <name>oozie.launcher.mapreduce.user.classpath.first</name>
+                    <value>true</value>
+                </property>
+            </configuration>
+            <main-class>eu.dnetlib.dhp.aggregation.mdstore.MDStoreActionNode</main-class>
+            <arg>--action</arg><arg>ROLLBACK</arg>
+            <arg>--mdStoreVersion</arg><arg>${wf:actionData('StartTransaction')['mdStoreVersion']}</arg>
+            <arg>--mdStoreManagerURI</arg><arg>${mdStoreManagerURI}</arg>
+        </java>
+        <ok to="Kill"/>
+        <error to="Kill"/>
+    </action>
+
     <end name="End"/>
+
 </workflow-app>

dhp-workflows/dhp-aggregation/src/main/scala/eu/dnetlib/dhp/sx/bio/SparkTransformBioDatabaseToOAF.scala

@@ -2,13 +2,15 @@ package eu.dnetlib.dhp.sx.bio
 
 import eu.dnetlib.dhp.application.ArgumentApplicationParser
 import eu.dnetlib.dhp.collection.CollectionUtils
+import eu.dnetlib.dhp.common.Constants.{MDSTORE_DATA_PATH,MDSTORE_SIZE_PATH}
+import eu.dnetlib.dhp.schema.mdstore.MDStoreVersion
 import eu.dnetlib.dhp.schema.oaf.Oaf
 import eu.dnetlib.dhp.sx.bio.BioDBToOAF.ScholixResolved
 import org.apache.commons.io.IOUtils
 import org.apache.spark.SparkConf
 import org.apache.spark.sql.{Encoder, Encoders, SparkSession}
 import org.slf4j.{Logger, LoggerFactory}
-
+import eu.dnetlib.dhp.utils.DHPUtils.{MAPPER, writeHdfsFile}
 object SparkTransformBioDatabaseToOAF {
 
   def main(args: Array[String]): Unit = {
@@ -25,8 +27,13 @@ object SparkTransformBioDatabaseToOAF {
 
     val dbPath: String = parser.get("dbPath")
     log.info("dbPath: {}", database)
-    val targetPath: String = parser.get("targetPath")
+
-    log.info("targetPath: {}", database)
+    val mdstoreOutputVersion = parser.get("mdstoreOutputVersion")
+    log.info("mdstoreOutputVersion: {}", mdstoreOutputVersion)
+
+    val cleanedMdStoreVersion = MAPPER.readValue(mdstoreOutputVersion, classOf[MDStoreVersion])
+    val outputBasePath = cleanedMdStoreVersion.getHdfsPath
+    log.info("outputBasePath: {}", outputBasePath)
 
     val spark: SparkSession =
       SparkSession
@@ -43,24 +50,28 @@ object SparkTransformBioDatabaseToOAF {
       case "UNIPROT" =>
         CollectionUtils.saveDataset(
           spark.createDataset(sc.textFile(dbPath).flatMap(i => BioDBToOAF.uniprotToOAF(i))),
-          targetPath
+          s"$outputBasePath/$MDSTORE_DATA_PATH"
         )
       case "PDB" =>
         CollectionUtils.saveDataset(
           spark.createDataset(sc.textFile(dbPath).flatMap(i => BioDBToOAF.pdbTOOaf(i))),
-          targetPath
+          s"$outputBasePath/$MDSTORE_DATA_PATH"
         )
       case "SCHOLIX" =>
         CollectionUtils.saveDataset(
           spark.read.load(dbPath).as[ScholixResolved].map(i => BioDBToOAF.scholixResolvedToOAF(i)),
-          targetPath
+          s"$outputBasePath/$MDSTORE_DATA_PATH"
         )
       case "CROSSREF_LINKS" =>
         CollectionUtils.saveDataset(
           spark.createDataset(sc.textFile(dbPath).map(i => BioDBToOAF.crossrefLinksToOaf(i))),
-          targetPath
+          s"$outputBasePath/$MDSTORE_DATA_PATH"
         )
     }
+
+    val df = spark.read.text(s"$outputBasePath/$MDSTORE_DATA_PATH")
+    val mdStoreSize = df.count
+    writeHdfsFile(spark.sparkContext.hadoopConfiguration, s"$mdStoreSize", s"$outputBasePath/$MDSTORE_SIZE_PATH")
   }
 
 }

dhp-workflows/dhp-aggregation/src/main/scala/eu/dnetlib/dhp/sx/bio/ebi/SparkEBILinksToOaf.scala

@@ -9,6 +9,9 @@ import org.apache.commons.io.IOUtils
 import org.apache.spark.SparkConf
 import org.apache.spark.sql._
 import org.slf4j.{Logger, LoggerFactory}
+import eu.dnetlib.dhp.common.Constants.{MDSTORE_DATA_PATH, MDSTORE_SIZE_PATH}
+import eu.dnetlib.dhp.schema.mdstore.MDStoreVersion
+import eu.dnetlib.dhp.utils.DHPUtils.{MAPPER, writeHdfsFile}
 
 object SparkEBILinksToOaf {
 
@@ -32,8 +35,13 @@ object SparkEBILinksToOaf {
     import spark.implicits._
     val sourcePath = parser.get("sourcePath")
     log.info(s"sourcePath  -> $sourcePath")
-    val targetPath = parser.get("targetPath")
+    val mdstoreOutputVersion = parser.get("mdstoreOutputVersion")
-    log.info(s"targetPath  -> $targetPath")
+    log.info("mdstoreOutputVersion: {}", mdstoreOutputVersion)
+
+    val cleanedMdStoreVersion = MAPPER.readValue(mdstoreOutputVersion, classOf[MDStoreVersion])
+    val outputBasePath = cleanedMdStoreVersion.getHdfsPath
+    log.info("outputBasePath: {}", outputBasePath)
+
     implicit val PMEncoder: Encoder[Oaf] = Encoders.kryo(classOf[Oaf])
 
     val ebLinks: Dataset[EBILinkItem] = spark.read
@@ -46,7 +54,10 @@ object SparkEBILinksToOaf {
         .flatMap(j => BioDBToOAF.parse_ebi_links(j.links))
         .filter(p => BioDBToOAF.EBITargetLinksFilter(p))
         .flatMap(p => BioDBToOAF.convertEBILinksToOaf(p)),
-      targetPath
+      s"$outputBasePath/$MDSTORE_DATA_PATH"
     )
+    val df = spark.read.text(s"$outputBasePath/$MDSTORE_DATA_PATH")
+    val mdStoreSize = df.count
+    writeHdfsFile(spark.sparkContext.hadoopConfiguration, s"$mdStoreSize", s"$outputBasePath/$MDSTORE_SIZE_PATH")
   }
 }