diff --git a/workflow/dnet/orcid_propagate.py b/workflow/dnet/orcid_propagate.py new file mode 100644 index 0000000..f8b73dd --- /dev/null +++ b/workflow/dnet/orcid_propagate.py @@ -0,0 +1,55 @@ +import os +from datetime import timedelta + +from airflow.decorators import dag +from airflow.models.param import Param +from airflow.providers.cncf.kubernetes.operators.spark_kubernetes import SparkKubernetesOperator + +from spark_configurator import SparkConfigurator + +EXECUTION_TIMEOUT = int(os.getenv("EXECUTION_TIMEOUT", 6)) + +default_args = { + "execution_timeout": timedelta(days=EXECUTION_TIMEOUT), + "retries": int(os.getenv("DEFAULT_TASK_RETRIES", 1)), + "retry_delay": timedelta(seconds=int(os.getenv("DEFAULT_RETRY_DELAY_SECONDS", 60))) +} + + +@dag( + dag_id="orcid_propagation_graph", + dag_display_name="Propagate ORCID data in graph", + default_args=default_args, + params={ + "S3_CONN_ID": Param("s3_conn", type='string', description="Airflow connection of S3 endpoint"), + "ORCID_PATH": Param("s3a://graph/tmp/prod_provision/graph/09_graph_orcid_enriched", type='string', description=""), + "INPUT_PATH": Param("s3a://graph/tmp/prod_provision/graph/09_graph_orcid_enriched", type='string', description=""), + "OUTPUT_PATH": Param("s3a://graph/tmp/prod_provision/graph/10_graph_propagated", type='string', + description=""), + "WRKDIR_PATH": Param("s3a://graph/tmp/prod_provision/working_dir/orcid_propagation", type='string', + description="") + }, + tags=["openaire"] +) +def orcid_propagation_dag(): + orcid_propagate = SparkKubernetesOperator( + task_id='PropagateGraphWithOrcid', + task_display_name="Propagate ORCID data", + namespace='dnet-spark-jobs', + template_spec=SparkConfigurator( + name="orcidpropagate-{{ ds }}-{{ task_instance.try_number }}", + mainClass="eu.dnetlib.dhp.orcidtoresultfromsemrel.SparkPropagateOrcidAuthor", + jarLocation='s3a://binaries/dhp-shade-package-1.2.5-SNAPSHOT.jar', + arguments=["--orcidPath", "{{ dag_run.conf.get('ORCID_PATH') }}", + "--graphPath", "{{ dag_run.conf.get('INPUT_PATH') }}", + "--targetPath", "{{ dag_run.conf.get('OUTPUT_PATH') }}", + "--workingDir", "{{ dag_run.conf.get('WRKDIR_PATH') }}", + "--matchingSource", "graph" + ]).get_configuration(), + kubernetes_conn_id="kubernetes_default" + ) + + orcid_propagate + + +orcid_propagation_dag()