Airflow BranchPythonOperator 不遵循指定的分支

Question

我有一个具有以下结构的 AIRFLOW DAG。

所有以“check*”开头的函数都是BranchPythonOperator，函数exceptionControl是一个ExecuteDagRunOperator，它接收每个错误以便处理它们。

这是 DAG 配置：

checkCloudFunctions = BranchPythonOperator(
    task_id='checkCloudFunctions',
    python_callable=check_cloud_functions,
    provide_context=True,
    dag=dag)

checkSqlTables = BranchPythonOperator(
    task_id='checkSqlTables',
    python_callable=check_sql_tables,
    provide_context=True,
    dag=dag)

checkBigQueryTable = BranchPythonOperator(
    task_id='checkBigQueryTable',
    python_callable=check_big_query_table,
    provide_context=True,
    dag=dag)

labBuilt = DummyOperator(
    task_id='labBuilt',
    dag=dag)

exceptionControl = ExecuteDagRunOperator(
    task_id='exceptionControl',
    execute_dag_id="SYS_exception_control",
    python_callable=mediation.dag_trigger_exception,
    trigger_rule='one_success',
    dag=dag)

# graphs
checkCloudFunctions >> checkSqlTables
checkCloudFunctions >> exceptionControl

checkSqlTables >> checkBigQueryTable
checkSqlTables >> exceptionControl

checkBigQueryTable >> labBuilt
checkBigQueryTable >> exceptionControl

问题是 checkSqlTables 应该遵循异常控制，但它会跳过并且 DAG 结束。该函数返回“exceptionControl”，我们可以在 checkSqlTables 日志中看到：

   {base_task_runner.py:98} INFO - {python_operator.py:90} INFO - Done. Returned value was: exceptionControl
   {base_task_runner.py:98} INFO - {python_operator.py:118} INFO - Following branch exceptionControl
   {base_task_runner.py:98} INFO - {python_operator.py:119} INFO - Marking other directly downstream tasks as skipped
   {base_task_runner.py:98} INFO - {python_operator.py:128} INFO - Done.

我也玩过 trigger_rule 属性（one_success、dummy...），但它似乎不起作用。

如果我删除第一步，它似乎可以工作，所以它似乎应该是我的dag的某种配置问题。

任何想法为什么函数 checkSqlTables 不分支到 exceptionControl？

编辑：在new deep reading to the Airflow Documentation 中，我注意到如果一个步骤将任务标记为已跳过，它将永远被跳过，因此我的代码将永远无法与分支运算符一起使用。

使用分支的解决方案包括在每一步之前的一个虚拟步骤。但是我有一些有超过 10 个步骤的 DAG，架构会完全混乱。

Answer 1

这个问题涉及气流如何标记任务的状态。 exceptionControl 将被屏蔽为跳过，而 check* 任务为 True。由于有多个check* 任务，第一次之后的check* 将无法更新exceptionControl 的状态，因为它已被屏蔽为跳过。

要解决这个问题，您可以在每个check* 任务中创建一个虚拟运算符。例如

         check1
           ^
        __.   _____________
       |                  |
    check2               dummy
      |                   |
     ...                exceptionControl

这似乎是这里讨论的一个错误https://github.com/apache/airflow/issues/10725

基于以上讨论，这里合并了一个修复https://github.com/apache/airflow/pull/11120。