使用external_task_sensor和external_sensor模块。

示例代码：

from airflow import DAG from airflow.operators.python_operator import PythonOperator from airflow.operators.bash_operator import BashOperator from airflow.sensors.external_task_sensor import ExternalTaskSensor from airflow.sensors.external_sensor import ExternalSensor from datetime import datetime, timedelta import os

default_args = { 'owner': 'airflow', 'depends_on_past': False, 'start_date': datetime(2021, 9, 1), 'email': ['airflow@example.com'], 'email_on_failure': False, 'email_on_retry': False, 'retries': 1, 'retry_delay': timedelta(minutes=5), }

dag = DAG( 'external_task_and_sensor_example', default_args=default_args, description='External sensor and external task sensor example', schedule_interval=timedelta(days=1), )

t1 = BashOperator( task_id='create_file', bash_command='echo "hello" > /tmp/test_file.txt', dag=dag, )

t2 = ExternalTaskSensor( task_id='wait_for_t1', external_dag_id='external_dag', external_task_id='create_file', mode='reschedule', dag=dag, )

t3 = ExternalSensor( task_id='wait_for_file', external_dag_id=None, external_task_id=None, external_sensor_id='file_sensor', external_system='file', external_system_args={'filepath': '/tmp/test_file.txt'}, mode='reschedule', dag=dag, )

t4 = PythonOperator( task_id='delete_file', python_callable=lambda: os.remove('/tmp/test_file.txt'), dag=dag, )

t2 >> t3 >> t4

注：上述代码中，external_dag由外部dag创建并传递。创建文件任务的外部任务ID为“create_file”，传递给wait_for_t1 external_task_id。等待文件传感器的external_sensor_id为“file_sensor”，且它使用filepath参数检查文件是否存在。