打破在气流中传播到下游的跳过状态答案

【问题标题】：Breaking the Skipped State being propagated to downstream in Airflow打破在气流中传播到下游的跳过状态
【发布时间】：2020-09-14 16:47:38
【问题描述】：

我有以下场景/DAG；

             |----->Task1----|                  |---->Task3---|
start task-->|               |-->Merge Task --->|             | ----->End Task
             |----->Task2----|                  |---->Task4---|

目前Task、Task2、Task3和Task4都是ShortCircuitOperators，当Task1和Task2其中之一被ShortCircuit时，所有下游任务都会被跳过。

但我的要求是打破在合并任务中传播到 Task3 和 Task4 的跳过状态。因为无论上游发生什么，我都希望任务 3 和任务 4 运行。有没有办法我可以做到这一点。？我希望在 DAG 中描述/显示依赖项。

【问题讨论】：

标签： airflow

【解决方案1】：

是的可以实现

- 不使用ShortCircuitOperator，而是使用AirflowSkipException（在PythonOperator 内）跳过任务（即有条件地执行任务/分支）
- 您也许可以使用BranchPythonOperator 实现同样的目的
- 但ShortCircuitOperator 的行为绝对不像大多数people's expectations。引用这一行与this 链接中的问题非常相似
  
  ...当上游之一被 ShortCircuitOperator 跳过时此任务也被跳过。我不希望最后的任务被跳过因为它必须报告 DAG 成功。
  
  为了避免它被跳过，我使用了 trigger_rule='all_done'，但它仍然被跳过。
  
  如果我使用 BranchPythonOperator 而不是 ShortCircuitOperator final 任务不会被跳过。 ...
- 此外，docs 会警告我们（这确实是ShortCircuitOperator 的预期行为）
  
  它评估一个条件，如果条件为 False，则使工作流程短路。任何下游任务都标有状态 “跳过”。
- 对于您（可能）跳过任务的下游任务，请使用不同的trigger_rules
- 所以不要使用默认的all_success，而是使用none_failed 或all_done 之类的东西（取决于您的要求）

【讨论】：

我试图了解气流在哪里为所有 dag 下游任务设置“跳过”状态。我在 skipmixin.py 中看到过代码，看起来它正在为直接下游任务设置“跳过”状态。我可以创建一个自定义 ShortCircuitOperator，它不会根据 task_id 模式或 Operator Type 将“跳过”状态传播/标记到一堆任务吗？