当上游跳过时气流“none_failed”跳过

Question

我有一个工作流程，其中有两个并行进程 (sentinel_run和sentinel_skip)，它们应该根据条件运行或跳过，然后连接在一起 ( resolve)。我需要直接位于任一sentinel_任务下游的任务进行级联跳过，但是当它到达该resolve任务时，resolve应该运行，除非上游任一进程出现故障。

根据文档，“none_failed”触发规则应该有效：

none_failed：所有父级都没有失败（失败或upstream_failed），即所有父级都已成功或被跳过

这也是对相关问题的回答。

然而，当我实现一个简单的例子时，我看到的并不是这样：

from airflow.models import DAG
from airflow.operators.dummy_operator import DummyOperator
from airflow.operators.python_operator import ShortCircuitOperator
from airflow.utils.dates import days_ago

dag = DAG(
    "testing",
    catchup=False,
    schedule_interval="30 12 * * *",
    default_args={
        "owner": "test@gmail.com",
        "start_date": days_ago(1),
        "catchup": False,
        "retries": 0
    }
)

start = DummyOperator(task_id="start", dag=dag)

sentinel_run = ShortCircuitOperator(task_id="sentinel_run", dag=dag, python_callable=lambda: True)
sentinel_skip = ShortCircuitOperator(task_id="sentinel_skip", dag=dag, python_callable=lambda: False)

a = DummyOperator(task_id="a", dag=dag)
b = DummyOperator(task_id="b", dag=dag)
c = DummyOperator(task_id="c", dag=dag)
d = DummyOperator(task_id="d", dag=dag)
e = DummyOperator(task_id="e", dag=dag)
f = DummyOperator(task_id="f", dag=dag)
g = DummyOperator(task_id="g", dag=dag)

resolve = DummyOperator(task_id="resolve", dag=dag, trigger_rule="none_failed")

start >> sentinel_run >> a >> b >> c >> resolve
start >> sentinel_skip >> d >> e >> f >> resolve

resolve >> g

此代码创建以下 dag：

问题是该resolved任务应该执行（因为上游没有任何一个upstream_failed或failed），但它正在跳过。

我已经检查了数据库，并且没有隐藏任何失败或上游失败的任务，而且我不明白为什么它不遵守“none_failed”逻辑。

我知道“丑陋的解决方法”，并已在其他工作流程中实现了它，但它增加了另一个要执行的任务，并增加了 DAG 的新用户必须理解的复杂性（特别是当您将其乘以多个任务时......） 。这是我从 Airflow 1.8 升级到 Airflow 1.10 的主要原因，所以我希望我缺少一些明显的东西......

Answer 1

记录下来是因为这个问题已经困扰了我两次，现在我已经解决了两次。

问题分析

当您将日志级别设置为 DEBUG 时，您将开始看到发生了什么：

[2019-10-09 18:30:05,472] {python_operator.py:114} INFO - Done. Returned value was: False
[2019-10-09 18:30:05,472] {python_operator.py:159} INFO - Condition result is False
[2019-10-09 18:30:05,472] {python_operator.py:165} INFO - Skipping downstream tasks...
[2019-10-09 18:30:05,472] {python_operator.py:168} DEBUG - Downstream task_ids [<Task(DummyOperator): f>, <Task(DummyOperator): g>, <Task(DummyOperator): d>, <Task(DummyOperator): resolve>, <Task(DummyOperator): e>]
[2019-10-09 18:30:05,492] {python_operator.py:173} INFO - Done.

由此，您可以看到问题不是“none_failed”错误地处理任务，而是模拟跳过条件的哨兵将所有下游依赖项标记为直接跳过。这是 ShortCircuitOperator 的行为- 跳过所有下游，包括下游任务的下游任务。

解决方案

此问题的解决方案在于认识到导致问题的是 ShortCircuitOperator 的行为，而不是 TriggerRule。一旦我们意识到这一点，就该开始编写一个更适合我们实际想要完成的任务的运算符了。

我已经包含了我当前使用的运算符；我欢迎任何有关处理单个下游任务修改的更好方法的意见。我确信有一个更好的习惯用法“跳过下一个，让其余的根据它们的触发规则级联”，但我已经在这方面花费了比我想要的更多的时间，并且我怀疑答案更深层次地存在于内部结构。

"""Sentinel Operator Plugin"""

import datetime

from airflow import settings
from airflow.models import SkipMixin, TaskInstance
from airflow.operators.python_operator import PythonOperator
from airflow.plugins_manager import AirflowPlugin
from airflow.utils.state import State


class SentinelOperator(PythonOperator, SkipMixin):
    """
    Allows a workflow to continue only if a condition is met. Otherwise, the
    workflow skips cascading downstream to the next time a viable task
    is identified.

    The SentinelOperator is derived from the PythonOperator. It evaluates a
    condition and stops the workflow if the condition is False. Immediate
    downstream tasks are skipped. If the condition is True, downstream tasks
    proceed as normal.

    The condition is determined by the result of `python_callable`.
    """
    def execute(self, context):
        condition = super(SentinelOperator, self).execute(context)
        self.log.info("Condition result is %s", condition)

        if condition:
            self.log.info('Proceeding with downstream tasks...')
            return

        self.log.info('Skipping downstream tasks...')

        session = settings.Session()

        for task in context['task'].downstream_list:
            ti = TaskInstance(task, execution_date=context['ti'].execution_date)
            self.log.info('Skipping task: %s', ti.task_id)
            ti.state = State.SKIPPED
            ti.start_date = datetime.datetime.now()
            ti.end_date = datetime.datetime.now()
            session.merge(ti)

        session.commit()
        session.close()

        self.log.info("Done.")


class Plugin_SentinelOperator(AirflowPlugin):
    name = "sentinel_operator"
    operators = [SentinelOperator]

经过修改，这会产生预期的 dag 结果：