Правильный способ создания динамических рабочих процессов в Airflow ⇐ Python

[Anonymous]

Проблема

Есть ли какой-нибудь способ в Airflow создать рабочий процесс, в котором количество задач B.* неизвестно до завершения задачи A? Я рассмотрел субдаги, но похоже, что они могут работать только со статическим набором задач, которые должны быть определены при создании дага.

Будут ли работать триггеры дага? Если да, то не могли бы вы привести пример?

У меня возникла проблема: невозможно узнать количество задач B, которые потребуются для расчета задачи C, пока задача A не будет завершена. Вычисление каждой задачи B.* займет несколько часов и не может быть объединено.

Код: Выделить всё

              |---> Task B.1 --|
|---> Task B.2 --|
Task A ------|---> Task B.3 --|-----> Task C
|       ....     |
|---> Task B.N --|

Идея №1

Мне не нравится это решение, потому что мне нужно создать блокирующий внешний датчик задач, и выполнение всей задачи B.* займет от 2 до 24 часов. Поэтому я не считаю это жизнеспособным решением. Наверняка есть более простой способ? Или Airflow не предназначен для этого?

Код: Выделить всё

Dag 1
Task A -> TriggerDagRunOperator(Dag 2) -> ExternalTaskSensor(Dag 2, Task Dummy B) -> Task C

Dag 2 (Dynamically created DAG though python_callable in TriggerDagrunOperator)
|-- Task B.1 --|
|-- Task B.2 --|
Task Dummy A --|-- Task B.3 --|-----> Task Dummy B
|     ....     |
|-- Task B.N --|

Изменить 1:

На данный момент на этот вопрос все еще нет хорошего ответа. Ко мне обратились несколько человек в поисках решения.

Подробнее здесь: https://stackoverflow.com/questions/415 ... in-airflow