Воздушный поток: Как получить выходные данные одной задачи, чтобы установить зависимости запускаемых ниже задач? - PullRequest
Новая
       14

Воздушный поток: Как получить выходные данные одной задачи, чтобы установить зависимости запускаемых ниже задач?

0 голосов
/ 13 марта 2019

У нас есть оператор модуля kubernetes, который выдаст словарь python, который определит, какие дальнейшие операторы модуля kubernetes ниже по потоку будут запускаться вместе со своими зависимостями и переменными среды, которые будут передаваться в каждый оператор.

Как мневернуть этот объект словаря python обратно в контекст исполнителя (или это рабочий контекст?), чтобы поток воздуха мог порождать нижестоящие операторы kubernetes?

Я смотрел на BranchOperator и TriggerDagRunOperator и XCOM push / pull и Variable.get и Variable.set, но, похоже, ничего не работает.

1 Ответ

2 голосов
/ 13 марта 2019

У нас есть оператор модуля kubernetes, который выплюнет питона словарь, который определит, какой дальнейший вниз по течению стручок kubernetes операторы для запуска

Это возможно, хотя и не так, как вы пытаетесь. Вам нужно будет иметь все возможные KubernetesPodOperator s уже в вашем рабочем процессе, а затем пропустить те, которые не нужно запускать .

Элегантный способ сделать это - прикрепить ShortCircuitOperator перед каждым KubernetesPodOperator, читающим опубликованный XCom ( словарь ) восходящим потоком KubernetesPodOperator и определяет, следует ли продолжить выполнение нижестоящего задания.

EDIT-1

На самом деле более чистым способом было бы просто поднять AirflowSkipException внутри задачи, которую вы хотите пропустить (вместо использования отдельного ShortCircuitOperator для этого)

Как мне получить этот словарь Python ... чтобы поток воздуха мог порождать операторы нисходящего kubernetes ..

Нет. Вы не можете динамически создавать новые задачи , основываясь на результатах восходящей задачи .

Подумайте об этом так: для scheduler обязательно знать все задачи (их task_id с, trigger_rule с, priority_weight и т. Д.) Заблаговременно, чтобы иметь возможность выполнить их, когда подходящее время наступает Если бы задачи просто продолжали появляться динамически, то планировщик Airflow должен был бы стать похожим на планировщик Operating System (!). Для более подробной информации читайте РЕДАКТИРОВАТЬ-1 часть из это ответ

...