Airflow SubDagOperator 死锁

Question

我遇到了一个问题，即由多个 SubDagOperator 组成的 DAG 无限期挂起。

设置:
使用 CeleryExecutor。出于本示例的目的，假设我们有一个可以同时运行五个任务的工作人员。
我遇到问题的 DAG 并行运行多个 SubDagOperator。出于说明目的，请考虑下图，其中每个节点都是一个 SubDagOperator:


问题:DAG 将停止在 DAG 的高并行部分取得进展。根本原因似乎是顶级 SubDagOperators 占用了所有五个可用于运行任务的插槽，因此这些 SubDagOperators 中的子任务都无法运行。这些子任务陷入排队状态，没有人取得进展。

SubDagOperator 任务会与它们自己的子任务竞争任务运行插槽，这让我有点惊讶，但现在对我来说很有意义。是否有关于编写我违反的 SubDagOperator 的最佳实践？

我的计划是通过创建自定义运算符来封装当前封装在 SubDagOperators 中的任务来解决此问题。我想知道是否有人对创建由其他运算符组成的运算符是否可取有建议？

Answer 1

似乎应该避免使用 SubDagOperator，因为它会导致此死锁问题。我最终发现，对于我的用例，我最好编写自己的自定义 BaseOperator子类来完成我在 SubDagOperator 中所做的任务。编写运算符类比我预期的要容易得多。