TriggerDagRunOperator 的条件触发答案

【问题标题】：conditionally_trigger for TriggerDagRunOperatorTriggerDagRunOperator 的条件触发
【发布时间】：2022-11-19 02:59:06
【问题描述】：

我有 2 个 DAG：dag_a 和 dag_b (dag_a -> dag_b) dag_a 执行后，调用 TriggerDagRunOperator，启动 dag_b。问题是，当 dag_b 关闭（暂停）时，dag_a 的 TriggerDagRunOperator 在 dag_b 中创建计划运行，只要 dag_a 正在运行就会排队。将 dag_b 重新打开后，队列中的任务开始执行。我正在尝试为 TriggerDagRunOperator 找到一个解决方案，即一个 conditionally_trigger 函数，如果 dag_b 暂停（关闭），它将跳过 TriggerDagRunOperator 任务的执行。我怎样才能做到这一点？

【问题讨论】：

标签： python airflow

【解决方案1】：

你可以使用ShortCircuitOperator来执行/跳过下游的dag_b。然后，使用 Airflow Rest API（或 shell/CLI）确定 dag_b 是否已暂停。

dag_a = TriggerDagRunOperator(
    trigger_dag_id='dag_a',
    ...
)

pause_check = ShortCircuitOperator(
    task_id='pause_check',
    python_callable=is_dag_paused,
    op_kwargs={
        'dag_id': 'dag_b'
    }
)

dag_b = TriggerDagRunOperator(
    trigger_dag_id='dag_b',
    ...
)

dag_a >> pause_check >> dag_b

和is_dag_paused函数可以是这样的。（这里我使用 Rest API。）

def is_dag_paused(**kwargs):
    import requests
    from requests.auth import HTTPBasicAuth
    
    dag_id = kwargs['dag_id']
    res = requests.get(f'http://{airflow_host}/api/v1/dags/{dag_id}/details',
                       auth=HTTPBasicAuth('username', 'pasword'))  # The auth method could be different for you. 

    if res.status_code == 200:
        rjson = res.json()
        # if you return True, the downstream tasks will be executed
        # if False, it will be skipped
        return not rjson['is_paused']
    else:
        print('Error: ', res)
        exit(1)

【讨论】：