<workflow-app name="clean graph" xmlns="uri:oozie:workflow:0.5">

    <parameters>
        <property>
            <name>graphInputPath</name>
            <description>the input path to read graph content</description>
        </property>
        <property>
            <name>graphOutputPath</name>
            <description>the target path to store cleaned graph</description>
        </property>
        <property>
            <name>isLookupUrl</name>
            <description>the address of the lookUp service</description>
        </property>
        <property>
            <name>shouldCleanContext</name>
            <description>true if the context have to be cleaned</description>
        </property>
        <property>
            <name>contextId</name>
            <value>sobigdata</value>
            <description>It is the context id that should be removed from the result if the condition is matched.
            Now it is just sobigdata. In a futere implementation I plan to have the contextId as value in a json
            where to specify also the constraints that should be verified to remove the context from the result</description>
        </property>
        <property>
            <name>verifyParam</name>
            <value>gcube </value>
            <description>It is the constrint to be verified. This time is hardcoded as gcube and it is searched for in
            the title. If title starts with gcube than the context sobigdata will be removed by the result if present</description>
        </property>

        <property>
            <name>sparkDriverMemory</name>
            <description>memory for driver process</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>oozieActionShareLibForSpark2</name>
            <description>oozie action sharelib for spark 2.*</description>
        </property>
        <property>
            <name>spark2ExtraListeners</name>
            <value>com.cloudera.spark.lineage.NavigatorAppListener</value>
            <description>spark 2.* extra listeners classname</description>
        </property>
        <property>
            <name>spark2SqlQueryExecutionListeners</name>
            <value>com.cloudera.spark.lineage.NavigatorQueryListener</value>
            <description>spark 2.* sql query execution listeners classname</description>
        </property>
        <property>
            <name>spark2YarnHistoryServerAddress</name>
            <description>spark 2.* yarn history server address</description>
        </property>
        <property>
            <name>spark2EventLogDir</name>
            <description>spark 2.* event log dir location</description>
        </property>
    </parameters>

    <start to="fork_clean_graph"/>

    <kill name="Kill">
        <message>Action failed, error message[${wf:errorMessage(wf:lastErrorNode())}]</message>
    </kill>

    <fork name="fork_clean_graph">
        <path start="clean_publication"/>
        <path start="clean_dataset"/>
        <path start="clean_otherresearchproduct"/>
        <path start="clean_software"/>
        <path start="clean_datasource"/>
        <path start="clean_organization"/>
        <path start="clean_project"/>
        <path start="clean_relation"/>
    </fork>

    <action name="clean_publication">
        <spark xmlns="uri:oozie:spark-action:0.2">
            <master>yarn</master>
            <mode>cluster</mode>
            <name>Clean publications</name>
            <class>eu.dnetlib.dhp.oa.graph.clean.CleanGraphSparkJob</class>
            <jar>dhp-graph-mapper-${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.sql.shuffle.partitions=7680
            </spark-opts>
            <arg>--inputPath</arg><arg>${graphInputPath}/publication</arg>
            <arg>--outputPath</arg><arg>${graphOutputPath}/publication</arg>
            <arg>--graphTableClassName</arg><arg>eu.dnetlib.dhp.schema.oaf.Publication</arg>
            <arg>--isLookupUrl</arg><arg>${isLookupUrl}</arg>
        </spark>
        <ok to="wait_clean"/>
        <error to="Kill"/>
    </action>

    <action name="clean_dataset">
        <spark xmlns="uri:oozie:spark-action:0.2">
            <master>yarn</master>
            <mode>cluster</mode>
            <name>Clean datasets</name>
            <class>eu.dnetlib.dhp.oa.graph.clean.CleanGraphSparkJob</class>
            <jar>dhp-graph-mapper-${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.sql.shuffle.partitions=7680
            </spark-opts>
            <arg>--inputPath</arg><arg>${graphInputPath}/dataset</arg>
            <arg>--outputPath</arg><arg>${graphOutputPath}/dataset</arg>
            <arg>--graphTableClassName</arg><arg>eu.dnetlib.dhp.schema.oaf.Dataset</arg>
            <arg>--isLookupUrl</arg><arg>${isLookupUrl}</arg>
        </spark>
        <ok to="wait_clean"/>
        <error to="Kill"/>
    </action>

    <action name="clean_otherresearchproduct">
        <spark xmlns="uri:oozie:spark-action:0.2">
            <master>yarn</master>
            <mode>cluster</mode>
            <name>Clean otherresearchproducts</name>
            <class>eu.dnetlib.dhp.oa.graph.clean.CleanGraphSparkJob</class>
            <jar>dhp-graph-mapper-${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.sql.shuffle.partitions=7680
            </spark-opts>
            <arg>--inputPath</arg><arg>${graphInputPath}/otherresearchproduct</arg>
            <arg>--outputPath</arg><arg>${graphOutputPath}/otherresearchproduct</arg>
            <arg>--graphTableClassName</arg><arg>eu.dnetlib.dhp.schema.oaf.OtherResearchProduct</arg>
            <arg>--isLookupUrl</arg><arg>${isLookupUrl}</arg>
        </spark>
        <ok to="wait_clean"/>
        <error to="Kill"/>
    </action>

    <action name="clean_software">
        <spark xmlns="uri:oozie:spark-action:0.2">
            <master>yarn</master>
            <mode>cluster</mode>
            <name>Clean softwares</name>
            <class>eu.dnetlib.dhp.oa.graph.clean.CleanGraphSparkJob</class>
            <jar>dhp-graph-mapper-${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.sql.shuffle.partitions=7680
            </spark-opts>
            <arg>--inputPath</arg><arg>${graphInputPath}/software</arg>
            <arg>--outputPath</arg><arg>${graphOutputPath}/software</arg>
            <arg>--graphTableClassName</arg><arg>eu.dnetlib.dhp.schema.oaf.Software</arg>
            <arg>--isLookupUrl</arg><arg>${isLookupUrl}</arg>
        </spark>
        <ok to="wait_clean"/>
        <error to="Kill"/>
    </action>

    <action name="clean_datasource">
        <spark xmlns="uri:oozie:spark-action:0.2">
            <master>yarn</master>
            <mode>cluster</mode>
            <name>Clean datasources</name>
            <class>eu.dnetlib.dhp.oa.graph.clean.CleanGraphSparkJob</class>
            <jar>dhp-graph-mapper-${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.sql.shuffle.partitions=7680
            </spark-opts>
            <arg>--inputPath</arg><arg>${graphInputPath}/datasource</arg>
            <arg>--outputPath</arg><arg>${graphOutputPath}/datasource</arg>
            <arg>--graphTableClassName</arg><arg>eu.dnetlib.dhp.schema.oaf.Datasource</arg>
            <arg>--isLookupUrl</arg><arg>${isLookupUrl}</arg>
        </spark>
        <ok to="wait_clean"/>
        <error to="Kill"/>
    </action>

    <action name="clean_organization">
        <spark xmlns="uri:oozie:spark-action:0.2">
            <master>yarn</master>
            <mode>cluster</mode>
            <name>Clean organizations</name>
            <class>eu.dnetlib.dhp.oa.graph.clean.CleanGraphSparkJob</class>
            <jar>dhp-graph-mapper-${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.sql.shuffle.partitions=7680
            </spark-opts>
            <arg>--inputPath</arg><arg>${graphInputPath}/organization</arg>
            <arg>--outputPath</arg><arg>${graphOutputPath}/organization</arg>
            <arg>--graphTableClassName</arg><arg>eu.dnetlib.dhp.schema.oaf.Organization</arg>
            <arg>--isLookupUrl</arg><arg>${isLookupUrl}</arg>
        </spark>
        <ok to="wait_clean"/>
        <error to="Kill"/>
    </action>

    <action name="clean_project">
        <spark xmlns="uri:oozie:spark-action:0.2">
            <master>yarn</master>
            <mode>cluster</mode>
            <name>Clean projects</name>
            <class>eu.dnetlib.dhp.oa.graph.clean.CleanGraphSparkJob</class>
            <jar>dhp-graph-mapper-${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.sql.shuffle.partitions=7680
            </spark-opts>
            <arg>--inputPath</arg><arg>${graphInputPath}/project</arg>
            <arg>--outputPath</arg><arg>${graphOutputPath}/project</arg>
            <arg>--graphTableClassName</arg><arg>eu.dnetlib.dhp.schema.oaf.Project</arg>
            <arg>--isLookupUrl</arg><arg>${isLookupUrl}</arg>
        </spark>
        <ok to="wait_clean"/>
        <error to="Kill"/>
    </action>

    <action name="clean_relation">
        <spark xmlns="uri:oozie:spark-action:0.2">
            <master>yarn</master>
            <mode>cluster</mode>
            <name>Clean relations</name>
            <class>eu.dnetlib.dhp.oa.graph.clean.CleanGraphSparkJob</class>
            <jar>dhp-graph-mapper-${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.sql.shuffle.partitions=7680
            </spark-opts>
            <arg>--inputPath</arg><arg>${graphInputPath}/relation</arg>
            <arg>--outputPath</arg><arg>${graphOutputPath}/relation</arg>
            <arg>--graphTableClassName</arg><arg>eu.dnetlib.dhp.schema.oaf.Relation</arg>
            <arg>--isLookupUrl</arg><arg>${isLookupUrl}</arg>
        </spark>
        <ok to="wait_clean"/>
        <error to="Kill"/>
    </action>

    <join name="wait_clean" to="clean_context"/>

    <decision name="clean_context">
        <switch>
            <case to="fork_clean_context">${wf:conf('shouldCleanContext') eq true}</case>
            <default to="End"/>
        </switch>
    </decision>


    <fork name="fork_clean_context">
        <path start="clean_publication_context"/>
        <path start="clean_dataset_context"/>
        <path start="clean_otherresearchproduct_context"/>
        <path start="clean_software_context"/>
    </fork>
    <action name="clean_publication_context">
        <spark xmlns="uri:oozie:spark-action:0.2">
            <master>yarn</master>
            <mode>cluster</mode>
            <name>Clean publications context</name>
            <class>eu.dnetlib.dhp.oa.graph.clean.CleanContextSparkJob</class>
            <jar>dhp-graph-mapper-${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.sql.shuffle.partitions=7680
            </spark-opts>
            <arg>--inputPath</arg><arg>${graphOutputPath}/publication</arg>
            <arg>--graphTableClassName</arg><arg>eu.dnetlib.dhp.schema.oaf.Publication</arg>
            <arg>--workingPath</arg><arg>${workingDir}/working/publication</arg>
            <arg>--contextId</arg><arg>${contextId}</arg>
            <arg>--verifyParam</arg><arg>${verifyParam}</arg>
        </spark>
        <ok to="wait_clean_context"/>
        <error to="Kill"/>
    </action>

    <action name="clean_dataset_context">
        <spark xmlns="uri:oozie:spark-action:0.2">
            <master>yarn</master>
            <mode>cluster</mode>
            <name>Clean datasets Context</name>
            <class>eu.dnetlib.dhp.oa.graph.clean.CleanContextSparkJob</class>
            <jar>dhp-graph-mapper-${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.sql.shuffle.partitions=7680
            </spark-opts>
            <arg>--inputPath</arg><arg>${graphOutputPath}/dataset</arg>
            <arg>--graphTableClassName</arg><arg>eu.dnetlib.dhp.schema.oaf.Dataset</arg>
            <arg>--workingPath</arg><arg>${workingDir}/working/dataset</arg>
            <arg>--contextId</arg><arg>${contextId}</arg>
            <arg>--verifyParam</arg><arg>${verifyParam}</arg>
        </spark>
        <ok to="wait_clean_context"/>
        <error to="Kill"/>
    </action>

    <action name="clean_otherresearchproduct_context">
        <spark xmlns="uri:oozie:spark-action:0.2">
            <master>yarn</master>
            <mode>cluster</mode>
            <name>Clean otherresearchproducts context</name>
            <class>eu.dnetlib.dhp.oa.graph.clean.CleanContextSparkJob</class>
            <jar>dhp-graph-mapper-${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.sql.shuffle.partitions=7680
            </spark-opts>
            <arg>--inputPath</arg><arg>${graphOutputPath}/otherresearchproduct</arg>
            <arg>--graphTableClassName</arg><arg>eu.dnetlib.dhp.schema.oaf.OtherResearchProduct</arg>
            <arg>--workingPath</arg><arg>${workingDir}/working/otherresearchproduct</arg>
            <arg>--contextId</arg><arg>${contextId}</arg>
            <arg>--verifyParam</arg><arg>${verifyParam}</arg>
        </spark>
        <ok to="wait_clean_context"/>
        <error to="Kill"/>
    </action>

    <action name="clean_software_context">
        <spark xmlns="uri:oozie:spark-action:0.2">
            <master>yarn</master>
            <mode>cluster</mode>
            <name>Clean softwares context</name>
            <class>eu.dnetlib.dhp.oa.graph.clean.CleanContextSparkJob</class>
            <jar>dhp-graph-mapper-${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.sql.shuffle.partitions=7680
            </spark-opts>
            <arg>--inputPath</arg><arg>${graphOutputPath}/software</arg>
            <arg>--graphTableClassName</arg><arg>eu.dnetlib.dhp.schema.oaf.Software</arg>
            <arg>--workingPath</arg><arg>${workingDir}/working/software</arg>
            <arg>--contextId</arg><arg>${contextId}</arg>
            <arg>--verifyParam</arg><arg>${verifyParam}</arg>
        </spark>
        <ok to="wait_clean_context"/>
        <error to="Kill"/>
    </action>

    <join name="wait_clean_context" to="End"/>

    <end name="End"/>
</workflow-app>