关于Airflow跨DAG依赖总结

单个DAG中Task之间的依赖

这是最常见的Task之间的依赖，在DAG中有多种方式指定依赖关系

# 定义DAGdag = DAG(...)# 定义task_atask_a = BashOperator(...)# 定义task_btask_b = BashOperator(...)# 指定task依赖关系task_a >> task_btask_a.set_downstream(task_b)# or# task_b << task_a# task_b.set_upstream(task_a)

这里简单指定了同一个DAG中两个task的依赖关系，即task_b依赖task_a，task_b只有等待task_a执行完且成功才能执行，这里task_b是task_a的下游（downstream）

不同DAG中Task之间的依赖

假设有这样一个需求，DAG A和DAG B分别用于处理两种不同的业务，但是DAG B中的task_b只有在DAG A中的task_a执行完且成功的情况下才能执行。很显然此时单个DAG中task的关系依赖已经不能满足需求，这时可以引入一个特殊的Operator：ExternalTaskSensor

ExternalTaskSensor

一种特殊的Operator，用于嗅探DAG外部Task的执行状态，通俗来说，就是让该ExternalTaskSensor作为DAG中的一个Task，专门用来等待外部的Task，而需要做具体业务操作的Task则作为该ExternalTaskSensor的下游，所以，只有ExternalTaskSensor这个Task成功了，下游的Task才能执行，这样便实现了跨DAG依赖的需求

主要参数

external_dag_id(str) – 包含需要等待的外部Task的GAD id

external_task_id(str or None) – 需要等待的外部Task，如果为None则等待整个DAG（默认为None）

allowed_states(Iterable) – DAG或Task允许的状态, 默认是 [‘success’]

failed_states(Iterable) –DAG或Task不允许的状态, 默认是 None

execution_delta(Optional[datetime.timedelta]) – 与之前执行的DAG或Task的时间差。这里的意思是指需要等待的DAG或Task的执行时间在当前DAG或Task的执行时间之前，也就是说在当前DAG或Task执行的时候需要等待的DAG活Task至少已经开始执行。 默认值是当前DAG或Task的执行时间. 如果需要指定前一天的DAG或Task，可以使用datetime.timedelta(days=1). execution_delta和execution_date_fn都可以作为ExternalTaskSensor的参数，但是不能同时使用.

execution_date_fn(Optional[Callable]) – 这是一个以当前执行时间作为参数的方法，用来返回期望的执行时间.通俗来说就是需要等待的DAG或Task执行时间在当前DAG的执行时间之后，用来往后倒一段时间，在该时间点的DAG或Task执行成功之前，该Task会一直等待，直到改时间点有DAG或Task执行成功或当前Task超时execution_delta和execution_date_fn都可以作为ExternalTaskSensor的参数，但是不能同时使用.

check_existence(bool) – 用于检查外部DAG id或者Task id是否存在（当为true时），如果不存在，则立刻停止等待

timeout(int) – 超时时间，单位为秒，如果在超时时间内还未等到外部DAG或Task成功执行，则抛出失败异常，进入重试（如果有重试次数的话）

mode(str) – 指定Task在等待期间的模式，有三种模式：

poke(默认): 在等待期间，Task会一直占用Worker Slotreschedule: Task只会在检查外部DAG或Task时才会占用一个Worker Slot，在两次检查之间会进入sleep状态（ExternalTaskSensor会每隔一段时间检查外部DAG或Task是否已执行成功，默认检查间隔时1分钟）smartsensor: 可进行批量处理的一种模式，内容较为复杂，具体配置可以参看这里Smart Sensors.

显然，在这里推荐使用reschedule模式。

话不多说，上案例

假如现在有两个DAG：mon.1d、handle.data.1d，其中mon.1d用于初始化所有DAG运行需要的环境参数，比如DAG需要运行在哪个环境（test or dev）再将参数保存到MySQL数据库，执行规则是0 1 * * *，也就是每天凌晨1点；而handle.data.1d需要先从MySQL数据库读取环境信息，再处理数据，执行规则是5 1 * * *，也就是每天凌晨1点05分。很显然，handle.data.1d需要等待mon.1d执行完后才能执行。这里虽然handle.data.1d的执行时间在mon.1d之后，但是我们不知道mon.1d具体需要执行多久才能执行完，万一因为环境或者网络因素导致DAG的执行时间超过5分钟，那handle.data.1d在1点05分执行时肯定会失败，所以handle.data.1d需要加一个ExterTaskSensor用来嗅探mon.1d是否在凌晨1点成功执行完毕。

该DAG中有四个Task： info_bash：输出一些当前DAG的信息，可以与DAG中其他Task并行init_env_conf：初始化环境参数信息save_env_conf：保存当前环境参数信息到数据库run_init_common_data：提交一个Spark任务，该任务基于当前环境参数运行

很显然，重要的是save_env_conf这个Task，只有该Task成功执行完毕，当前DAG中的Task和其他DAG才能执行

from datetime import datetime, timedeltafrom airflow import DAGfrom airflow.operators.bash import BashOperatorfrom airflow.operators.python import PythonOperatordefault_args = {'owner': 'example','email': ['example@'],'email_on_failure': True,'email_on_retry': True,'retries': 1,'retry_delay': timedelta(minutes=2),'start_date': datetime(, 12, 6),}common_dag = DAG(dag_id='mon.1d',default_args=default_args,description='init common data',schedule_interval='0 1 * * *',)run_info_bash = BashOperator(task_id='info_bash',bash_command='python3 -V', # 输出Python环境信息dag= common_dag,)run_init_env_conf = PythonOperator(task_id="init_env_conf",python_callable=init_env_conf, # 调用Python方法，初始化环境参数dag= common_dag,)run_save_env_conf = PythonOperator(task_id='save_env_conf',python_callable=save_env_conf, # 调用Python方法，保存环境参数到数据库dag=common_dag,)run_init_common_data = BashOperator(task_id='init_common_data',bash_command='spark-submit .....', # 提交spark任务dag= common_dag,)run_info_bashrun_init_env >> run_save_env >> run_init_common_data

handle.data.1d

在该DAG中有四个Task： run_info_bash：同上，输出DAG运行信息monitor_common_dag_save_env_conf：嗅探外部Task的Sensor，这里为嗅探即等待mon.1dDAG中的save_env_conf Taskrun_read_env_conf：读取环境信息run_handle_data：处理数据

from datetime import datetime, timedeltafrom airflow import DAGfrom airflow.operators.bash import BashOperatorfrom airflow.operators.python import PythonOperatorfrom airflow.sensors.external_task import ExternalTaskSensordefault_args = {'owner': 'example','email': ['example@'],'email_on_failure': True,'email_on_retry': True,'retries': 1,'retry_delay': timedelta(minutes=2),'start_date': datetime(, 12, 6),}handle_data_dag = DAG(dag_id='handle.data.1d',default_args=default_args,description='Handle Data',schedule_interval='5 1 * * *')run_info_bash = BashOperator(task_id='info_bash',bash_command='python3 -V', # 输出Python环境信息dag=handle_data_dag,)# 这里实例化一个ExterTaskSensormonitor_common_dag_save_env_conf = ExternalTaskSensor(task_id='monitor_common_dag_save_env_conf',external_dag_id='mon.1d', # 需要等待的外部DAG idexternal_task_id='save_env_conf', # 需要等待的外部Task idexecution_delta=timedelta(minutes=5), # 执行时间差，这里指定5分钟，那么当前ExternalTaskSensor会基于当前执行时间（1:05）往前倒5分钟（1:00）寻找在这个时间点已经成功执行完毕的**mon.1d**的save_env_conf## 假如「**mon.1d**」的执行规则是「10 1 * * *」也就是每天凌晨1点10分，## 那么这里可以使用「execution_date_fn」，让当前DAG等待至1点10分，## 直到「**mon.1d**」的「save_env_conf」成功执行完# execution_date_fn=lambda dt: dt + timedelta(minutes=5),timeout=600, # 超时时间，如果等待了600秒还未符合期望状态的外部Task，那么抛出异常进入重试allowed_states=['success'], # Task允许的状态，这里只允许外部Task执行状态为'success'mode='reschedule', # reschedule模式，在等待的时候，两次检查期间会sleep当前Task，节约系统开销check_existence=True,# 校验外部Task是否存在，不存在立马结束等待dag=handle_data_dag,)run_read_env_conf =PythonOperator(task_id="read_env_conf",python_callable=read_env_conf, # 读取环境信息dag=handle_data_dag,)run_handle_data = BashOperator(task_id='danhle_data',bash_command='spark-submit .....',dag=handle_data_dag,)run_info_bashmonitor_common_dag_save_env_conf >> run_read_env_conf >> run_handle_data

需要注意的地方

execution_delta和execution_date_fn关心的是DAG的Execution_Date，也就是DAG的「schedule_interval」所指定的执行时间，至于这个DAG什么时候执行完，execution_delta和execution_date_fn并不关心。其实通过Airflow的Tree View可以看出来，一次DAG的运行即是一次记录，如图。而DAG的运行成功与否的状态都会记录在此次运行记录中，execution_delta和execution_date_fn关心的便是图中框出来的两个地方

在Airflow中schedule_interval非常重要，因为在跨DAG依赖的场景中，这关乎到DAG能否正常成功运行，execution_delta和execution_date_fn两个参数都依赖于DAG的执行策略，所以，DAGs能否正常高效的运行，需要合理安排每个DAG的schedule_interval。