Как получить результат задачи и использовать его в потоке воздуха ⇐ Python

[Anonymous]

Я использую Airflow 2.10.2. Я определяю Dag, который в основном делает две вещи.

• Некоторая базовая логика, которая создает группу файлов
• Запускает параллельно (порядок не важен) набор «рабочих» задач

Поскольку первый шаг требует больших вычислительных ресурсов, я определяю его как Задача который возвращает количество созданных файлов. Это число важно, поскольку мне приходится запускать столько рабочих задач, сколько созданных файлов. Мой код выглядит примерно так:

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

@dag
def my_dag:

@task
def create_files():
...
return count

count = create_files()

for i in range(count):
worker.override()()

my_dag()

Я не понимаю, как мне получить значение count в моем определении дага. Вероятно, я просто не понимал, как работает Xcoms (так как в основном полагался на API TaskFlow), но буду рад любой помощи.
Я попробовал получить значение с помощью xcoms_pull< /code>:

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

count.xcoms_pull(task_ids='create_files')
count['ti'].xcoms_pull(task_ids='create_files')

И без xcoms_pull:

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

count['ti']

Но ни один из них не работает, поскольку XCom определены для взаимодействия между задачами.
Сообщение об ошибке:

< р>

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

AttributeError: 'XComArg' object count has no attribute 'xcoms_pull'

.

Теперь я увидел, что XCom — это всего лишь особый случай переменных, но документации по этому поводу практически нет (по крайней мере, эта Я нашел).

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