<workflow-app name="IRCDL Extention" xmlns="uri:oozie:workflow:0.5">


    <start to="deleteoutputpath"/>
    <kill name="Kill">
        <message>Action failed, error message[${wf:errorMessage(wf:lastErrorNode())}]</message>
    </kill>
    <action name="deleteoutputpath">
        <fs>
            <delete path='${outputPath}'/>
            <mkdir path='${outputPath}'/>
            <delete path='${workingDir}'/>
            <mkdir path='${workingDir}'/>
        </fs>
        <ok to="fork_prepare"/>
        <error to="Kill"/>
    </action>

    <fork name="fork_prepare">
        <path start="fork_prepare_result"/>
        <path start="prepare_orcid"/>
    </fork>


    <action name="prepare_orcid">
        <spark xmlns="uri:oozie:spark-action:0.2">
            <master>yarn</master>
            <mode>cluster</mode>
            <name>PrepareResult</name>
            <class>eu.dnetlib.dhp.ircdl_extention.PrepareNormalizedOrcid</class>
            <jar>dhp-aggregation-${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=3840
            </spark-opts>
            <arg>--inputPath</arg><arg>${orcidInputPath}</arg>
            <arg>--outputPath</arg><arg>${workingDir}/ORCID/entrySetMayNormalized/</arg>
        </spark>
        <ok to="join_fork"/>
        <error to="Kill"/>
    </action>

    <fork name="fork_prepare_result">
        <path start="prepare_publication"/>
        <path start="prepare_dataset"/>
        <path start="prepare_software"/>
        <path start="prepare_other"/>
    </fork>

    <action name="prepare_publication">
        <spark xmlns="uri:oozie:spark-action:0.2">
            <master>yarn</master>
            <mode>cluster</mode>
            <name>PrepareResult</name>
            <class>eu.dnetlib.dhp.ircdl_extention.PrepareResultSpark</class>
            <jar>dhp-aggregation-${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=3840
            </spark-opts>
            <arg>--inputPath</arg><arg>${inputPath}/publication</arg>
            <arg>--resultClass</arg><arg>eu.dnetlib.dhp.schema.oaf.Publication</arg>
            <arg>--outputPath</arg><arg>${workingDir}/GRAPH/publicationsWithOrcid</arg>

        </spark>
        <ok to="wait_prepare_result"/>
        <error to="Kill"/>
    </action>

    <action name="prepare_dataset">
        <spark xmlns="uri:oozie:spark-action:0.2">
            <master>yarn</master>
            <mode>cluster</mode>
            <name>PrepareResult</name>
            <class>eu.dnetlib.dhp.ircdl_extention.PrepareResultSpark</class>
            <jar>dhp-aggregation-${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=3840
            </spark-opts>
            <arg>--inputPath</arg><arg>${inputPath}/dataset</arg>
            <arg>--resultClass</arg><arg>eu.dnetlib.dhp.schema.oaf.Dataset</arg>
            <arg>--outputPath</arg><arg>${workingDir}/GRAPH/datasetWithOrcid</arg>

        </spark>
        <ok to="wait_prepare_result"/>
        <error to="Kill"/>
    </action>

    <action name="prepare_software">
        <spark xmlns="uri:oozie:spark-action:0.2">
            <master>yarn</master>
            <mode>cluster</mode>
            <name>PrepareResult</name>
            <class>eu.dnetlib.dhp.ircdl_extention.PrepareResultSpark</class>
            <jar>dhp-aggregation-${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=3840
            </spark-opts>
            <arg>--inputPath</arg><arg>${inputPath}/software</arg>
            <arg>--resultClass</arg><arg>eu.dnetlib.dhp.schema.oaf.Software</arg>
            <arg>--outputPath</arg><arg>${workingDir}/GRAPH/softwareWithOrcid</arg>

        </spark>
        <ok to="wait_prepare_result"/>
        <error to="Kill"/>
    </action>

    <action name="prepare_other">
        <spark xmlns="uri:oozie:spark-action:0.2">
            <master>yarn</master>
            <mode>cluster</mode>
            <name>PrepareResult</name>
            <class>eu.dnetlib.dhp.ircdl_extention.PrepareResultSpark</class>
            <jar>dhp-aggregation-${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=3840
            </spark-opts>
            <arg>--inputPath</arg><arg>${inputPath}/otherresearchproduct</arg>
            <arg>--resultClass</arg><arg>eu.dnetlib.dhp.schema.oaf.OtherResearchProduct</arg>
            <arg>--outputPath</arg><arg>${workingDir}/GRAPH/otherWithOrcid</arg>

        </spark>
        <ok to="wait_prepare_result"/>
        <error to="Kill"/>
    </action>


    <join name="wait_prepare_result" to="normalize_result"/>


    <action name="normalize_result">
        <spark xmlns="uri:oozie:spark-action:0.2">
            <master>yarn</master>
            <mode>cluster</mode>
            <name>PrepareResult</name>
            <class>eu.dnetlib.dhp.ircdl_extention.PrepareNormalizedResultSpark</class>
            <jar>dhp-aggregation-${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=3840
            </spark-opts>
            <arg>--inputPath</arg><arg>${workingDir}/GRAPH/</arg>
            <arg>--outputPath</arg><arg>${workingDir}/GRAPH/Normalized/</arg>
        </spark>
        <ok to="select_only_author_with_orcid"/>
        <error to="Kill"/>
    </action>


    <action name="select_only_author_with_orcid">
        <spark xmlns="uri:oozie:spark-action:0.2">
            <master>yarn</master>
            <mode>cluster</mode>
            <name>PrepareResult</name>
            <class>eu.dnetlib.dhp.ircdl_extention.SelectAuthorWithOrcidOnlySpark</class>
            <jar>dhp-aggregation-${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=3840
            </spark-opts>
            <arg>--inputPath</arg><arg>${workingDir}/GRAPH/Normalized/ResultAuthorNormalized/</arg>
            <arg>--outputPath</arg><arg>${workingDir}/GRAPH/Normalized/ResultWithOrcid/</arg>
        </spark>
        <ok to="fork_get_result_info"/>
        <error to="Kill"/>
    </action>

    <fork name="fork_get_result_info">
        <path start="get_result_instrepo"/>
        <path start="get_result_datacite"/>
        <path start="get_result_crossref"/>
    </fork>


    <action name="get_result_instrepo">
        <spark xmlns="uri:oozie:spark-action:0.2">
            <master>yarn</master>
            <mode>cluster</mode>
            <name>GetResultInstRepo</name>
            <class>eu.dnetlib.dhp.ircdl_extention.PrepareResultFromInstRepo</class>
            <jar>dhp-aggregation-${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=3840
            </spark-opts>
            <arg>--inputPath</arg><arg>${workingDir}/GRAPH/Normalized/ResultWithOrcid/</arg>
            <arg>--outputPath</arg><arg>${workingDir}/GRAPH/InstRepo/</arg>
            <arg>--datasourcePath</arg><arg>${datasourcePath}</arg>
        </spark>
        <ok to="wait_res_info"/>
        <error to="Kill"/>
    </action>

    <action name="get_result_datacite">
        <spark xmlns="uri:oozie:spark-action:0.2">
            <master>yarn</master>
            <mode>cluster</mode>
            <name>GetResultInstRepo</name>
            <class>eu.dnetlib.dhp.ircdl_extention.PrepareDataciteSpark</class>
            <jar>dhp-aggregation-${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=3840
            </spark-opts>
            <arg>--inputPath</arg><arg>${workingDir}/GRAPH/Normalized/ResultWithOrcid/</arg>
            <arg>--outputPath</arg><arg>${workingDir}/GRAPH/Datacite/</arg>
        </spark>
        <ok to="wait_res_info"/>
        <error to="Kill"/>
    </action>

    <action name="get_result_crossref">
        <spark xmlns="uri:oozie:spark-action:0.2">
            <master>yarn</master>
            <mode>cluster</mode>
            <name>GetResultInstRepo</name>
            <class>eu.dnetlib.dhp.ircdl_extention.PrepareCrossrefSpark</class>
            <jar>dhp-aggregation-${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=3840
            </spark-opts>
            <arg>--inputPath</arg><arg>${workingDir}/GRAPH/Normalized/ResultWithOrcid/</arg>
            <arg>--outputPath</arg><arg>${workingDir}/GRAPH/Crossref/</arg>
        </spark>
        <ok to="wait_res_info"/>
        <error to="Kill"/>
    </action>


    <join name="wait_res_info" to="get_result_alltherest"/>

    <action name="get_result_alltherest">
        <spark xmlns="uri:oozie:spark-action:0.2">
            <master>yarn</master>
            <mode>cluster</mode>
            <name>GetResultInstRepo</name>
            <class>eu.dnetlib.dhp.ircdl_extention.PrepareResultAllTheRestSpark</class>
            <jar>dhp-aggregation-${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=3840
            </spark-opts>
            <arg>--inputPath</arg><arg>${workingDir}/GRAPH/Normalized/ResultWithOrcid/</arg>
            <arg>--outputPath</arg><arg>${workingDir}/GRAPH/AllTheRest/</arg>
            <arg>--instRepoPath</arg><arg>${workingDir}/GRAPH/InstRepo/</arg>
            <arg>--datacitePath</arg><arg>${workingDir}/GRAPH/Datacite/</arg>
            <arg>--crossrefPath</arg><arg>${workingDir}/GRAPH/Crossref/</arg>
        </spark>
        <ok to="join_fork"/>
        <error to="Kill"/>
    </action>

    <join name="join_fork" to="fork_get_wrong"/>

    <fork name="fork_get_wrong">
        <path start="get_wrong_instrepo"/>
        <path start="get_wrong_datacite"/>
        <path start="get_wrong_crossref"/>
        <path start="get_wrong_alltherest"/>
        <path start="get_wrong_zenodo"/>
        <path start="get_wrong_figshare"/>
        <path start="get_wrong_dryad"/>
    </fork>

    <action name="get_wrong_instrepo">
        <spark xmlns="uri:oozie:spark-action:0.2">
            <master>yarn</master>
            <mode>cluster</mode>
            <name>GetResultInstRepo</name>
            <class>eu.dnetlib.dhp.ircdl_extention.WrongSpark</class>
            <jar>dhp-aggregation-${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=3840
            </spark-opts>
            <arg>--inputPath</arg><arg>${workingDir}/GRAPH/InstRepo/</arg>
            <arg>--outputPath</arg><arg>${outputPath}/InstRepo/</arg>
            <arg>--orcidPath</arg><arg>${workingDir}/ORCID/entrySetMayNormalized/</arg>
            <arg>--ap</arg><arg>${workingDir}/GRAPH/Normalized/ResultAuthorNormalized/</arg>
        </spark>
        <ok to="jojn_wrong"/>
        <error to="Kill"/>
    </action>
    <action name="get_wrong_datacite">
        <spark xmlns="uri:oozie:spark-action:0.2">
            <master>yarn</master>
            <mode>cluster</mode>
            <name>GetResultInstRepo</name>
            <class>eu.dnetlib.dhp.ircdl_extention.WrongSpark</class>
            <jar>dhp-aggregation-${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=3840
            </spark-opts>
            <arg>--inputPath</arg><arg>${workingDir}/GRAPH/Datacite/allDatacite/</arg>
            <arg>--outputPath</arg><arg>${outputPath}/Datacite/</arg>
            <arg>--orcidPath</arg><arg>${workingDir}/ORCID/entrySetMayNormalized/</arg>
            <arg>--ap</arg><arg>${workingDir}/GRAPH/Normalized/ResultAuthorNormalized/</arg>
        </spark>
        <ok to="jojn_wrong"/>
        <error to="Kill"/>
    </action>

    <action name="get_wrong_crossref">
        <spark xmlns="uri:oozie:spark-action:0.2">
            <master>yarn</master>
            <mode>cluster</mode>
            <name>GetResultInstRepo</name>
            <class>eu.dnetlib.dhp.ircdl_extention.WrongSpark</class>
            <jar>dhp-aggregation-${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=3840
            </spark-opts>
            <arg>--inputPath</arg><arg>${workingDir}/GRAPH/Crossref/</arg>
            <arg>--outputPath</arg><arg>${outputPath}/Crossref/</arg>
            <arg>--orcidPath</arg><arg>${workingDir}/ORCID/entrySetMayNormalized/</arg>
            <arg>--ap</arg><arg>${workingDir}/GRAPH/Normalized/ResultAuthorNormalized/</arg>
        </spark>
        <ok to="jojn_wrong"/>
        <error to="Kill"/>
    </action>

    <action name="get_wrong_alltherest">
        <spark xmlns="uri:oozie:spark-action:0.2">
            <master>yarn</master>
            <mode>cluster</mode>
            <name>GetResultInstRepo</name>
            <class>eu.dnetlib.dhp.ircdl_extention.WrongSpark</class>
            <jar>dhp-aggregation-${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=3840
            </spark-opts>
            <arg>--inputPath</arg><arg>${workingDir}/GRAPH/AllTheRest/</arg>
            <arg>--outputPath</arg><arg>${outputPath}/AllTheRest/</arg>
            <arg>--orcidPath</arg><arg>${workingDir}/ORCID/entrySetMayNormalized/</arg>
            <arg>--ap</arg><arg>${workingDir}/GRAPH/Normalized/ResultAuthorNormalized/</arg>
        </spark>
        <ok to="jojn_wrong"/>
        <error to="Kill"/>
    </action>

    <action name="get_wrong_zenodo">
        <spark xmlns="uri:oozie:spark-action:0.2">
            <master>yarn</master>
            <mode>cluster</mode>
            <name>GetResultInstRepo</name>
            <class>eu.dnetlib.dhp.ircdl_extention.WrongSpark</class>
            <jar>dhp-aggregation-${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=3840
            </spark-opts>
            <arg>--inputPath</arg><arg>${workingDir}/GRAPH/Datacite/Zenodo/</arg>
            <arg>--outputPath</arg><arg>${outputPath}/Zenodo/</arg>
            <arg>--orcidPath</arg><arg>${workingDir}/ORCID/entrySetMayNormalized/</arg>
            <arg>--ap</arg><arg>${workingDir}/GRAPH/Normalized/ResultAuthorNormalized/</arg>
        </spark>
        <ok to="jojn_wrong"/>
        <error to="Kill"/>
    </action>
    <action name="get_wrong_figshare">
        <spark xmlns="uri:oozie:spark-action:0.2">
            <master>yarn</master>
            <mode>cluster</mode>
            <name>GetResultInstRepo</name>
            <class>eu.dnetlib.dhp.ircdl_extention.WrongSpark</class>
            <jar>dhp-aggregation-${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=3840
            </spark-opts>
            <arg>--inputPath</arg><arg>${workingDir}/GRAPH/Datacite/Figshare/</arg>
            <arg>--outputPath</arg><arg>${outputPath}/Figshare/</arg>
            <arg>--orcidPath</arg><arg>${workingDir}/ORCID/entrySetMayNormalized/</arg>
            <arg>--ap</arg><arg>${workingDir}/GRAPH/Normalized/ResultAuthorNormalized/</arg>
        </spark>
        <ok to="jojn_wrong"/>
        <error to="Kill"/>
    </action>
    <action name="get_wrong_dryad">
        <spark xmlns="uri:oozie:spark-action:0.2">
            <master>yarn</master>
            <mode>cluster</mode>
            <name>GetResultInstRepo</name>
            <class>eu.dnetlib.dhp.ircdl_extention.WrongSpark</class>
            <jar>dhp-aggregation-${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=3840
            </spark-opts>
            <arg>--inputPath</arg><arg>${workingDir}/GRAPH/Datacite/Dryad/</arg>
            <arg>--outputPath</arg><arg>${outputPath}/Dryad/</arg>
            <arg>--orcidPath</arg><arg>${workingDir}/ORCID/entrySetMayNormalized/</arg>
            <arg>--ap</arg><arg>${workingDir}/GRAPH/Normalized/ResultAuthorNormalized/</arg>
        </spark>
        <ok to="jojn_wrong"/>
        <error to="Kill"/>
    </action>
    <join name="jojn_wrong" to="End"/>

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