エアフローSubDagOperatorデッドロック

Question

私はいくつかのSubDagOperatorsで構成されるDAGが無期限にハングする問題に遭遇しています。

セットアップ： CeleryExecutorの使用。この例では、5つのタスクを同時に実行できる1つのワーカーがいるとします。 問題に遭遇しているDAGは、複数のSubDagOperatorsを並列に実行します。説明のために、すべてのノードがSubDagOperatorである次のグラフを検討してください。

問題：DAGは、DAGの高並列部分での進行を停止します。根本的な原因は、トップレベルのSubDagOperatorsが実行中のタスクに使用できる5つのスロットすべてを占有しているため、それらのSubDagOperators内のいずれのサブタスクも実行できないようです。これらのサブタスクはキューに入れられた状態に固執し、誰も進歩しません。

SubDagOperatorタスクは、タスク実行スロットのための独自のサブタスクと競合しますが、今は私には間違いありません。私が違反しているSubDagOperatorsの作成に関するベストプラクティスはありますか？

私の計画は、現在、SubDagOperators内にカプセル化されているタスクをカプセル化するカスタム演算子を作成することでこの問題を回避することです。他の演算子で構成された演算子を作成することが賢明であるかどうかについて誰かが助言を持っていたかどうか疑問に思っていましたか？

Answer 1

このデッドロックの問題が発生するため、SubDagOperatorは避けてください。私は最終的に、私の使用事例では、SubDagOperator内でやっていた作業を行うために自分自身のカスタムBaseOperatorサブクラスを書くことで最善を尽くしたことを発見しました。演算子クラスを書くことは、私が予想していたよりずっと簡単でした。