Differential D2754
Run fantoir-datasource as Airflow pipeline Authored by dereckson on Jan 24 2023, 01:09. Tags None Referenced Files
Subscribers None
Details
Provide a _pipelines component for Apache Airflow The fantoir_fetch pipeline will run fantoir-datasource fetch. When a new version is available, the fetch pipeline will run the fantoir_import Pipelines are stored in the dags/ directory, Ref T1750 Run workflows
Diff Detail
Event Timeline
Comment Actions At the end of the TaskFlow DAG, function should be called, for example at the end of fantoir_fetch.py we need to add fantoir_fetch_dag(). Tested under Airflow 2.8.0, notes is an invalid argument. airflow@4c3c353e2f0c:/opt/airflow$ python /opt/airflow/dags/fantoir_fetch.py
Traceback (most recent call last):
File "/opt/airflow/dags/fantoir_fetch.py", line 113, in <module>
fantoir_fetch_dag()
File "/home/airflow/.local/lib/python3.11/site-packages/airflow/models/dag.py", line 3941, in factory
f(**f_kwargs)
File "/opt/airflow/dags/fantoir_fetch.py", line 77, in fantoir_fetch_dag
call_import_dag = TriggerDagRunOperator(
^^^^^^^^^^^^^^^^^^^^^^
File "/home/airflow/.local/lib/python3.11/site-packages/airflow/models/baseoperator.py", line 437, in apply_defaults
result = func(self, **kwargs, default_args=default_args)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/airflow/.local/lib/python3.11/site-packages/airflow/operators/trigger_dagrun.py", line 122, in __init__
super().__init__(**kwargs)
File "/home/airflow/.local/lib/python3.11/site-packages/airflow/models/baseoperator.py", line 437, in apply_defaults
result = func(self, **kwargs, default_args=default_args)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/airflow/.local/lib/python3.11/site-packages/airflow/models/baseoperator.py", line 798, in __init__
raise AirflowException(
airflow.exceptions.AirflowException: Invalid arguments were passed to TriggerDagRunOperator (task_id: call_import_dag). Invalid arguments were:
**kwargs: {'notes': 'Triggered by fantoir_fetch DAG, as a new version is available.'}Comment Actions TaskFlow DAGs are defined by a function, but that function needs to be called. As such, there are valid Python executables, hence the chmod 755. With that, there are correctly detected by Airflow on restart. Comment Actions
Comment Actions Currently, the fetch pipeline fails at the notify step: dag_run_id = task_instance.xcom_pull( task_id="call_import_dag", key="trigger_run_id" ) Probably task_ids as the code just above works, and XCOM variable is resolved according sentry. From https://sentry.nasqueron.org/organizations/airflow/issues/73: TypeError Comment Actions Notification successfully reached the Notifications Center. Here an extract of the event through Sentry discover:  It's been processed by Wearg on IRC and ignored, as group is unknown: 22:37:50 <Wearg> [DEBUG] Message for unknown group: Nasqueron Datasources | |||||||||||||||||||||||||||||||||||||||||||||||||||||||