🤝

Добрый день. Стоит max_active_runs=1. Запустили ДАГ вручную, он работал несколько часов, и параллельно сработал запуск по расписанию(шедулер) и сломал загрузку. Так и должно быть? Эта настройка не решает проблему?

А как себя вело в итоге? Запустился новый даг ран, а старый остановился?

Да, именно так произошло

Ну вообще логично: свежие данные в приоритете.
Есть пара гипотез/предложений:
– Попробовать запихнуть всё в очень маленький пул (не верю, что точно так сработает, но мб и не поставит новый)
– Добавить dagrun_timeout = разница между даг ранами - какое-то время, дагран старый упадет сам
– Гранулировать загрузку, смысл любого дага в том, чтобы он был идемпотентный (сколько раз не запускай даг ран за 1 января, результаты будут всё те же), если и упадёт – можно в конце самого даг рана находить предыдущий упавший и чистить его

Понял, спасибо большое. Буду решать проблему

привет, подскажите, туплю уже час
функция, которая берет время запуска дага и пушит его в xcom
def _time_ran(ti):
   time = '{{ ds }}'
   ti.xcom_push(key='now_time', value=time)

потом пишу такой питон оператор

run_time = PythonOperator(
   task_id='now_time',
   provide_context=True,
   python_callable=_time_ran,
   dag=dag
)

но в xcom пушится не время а просто {{ds}}

value =  {{ ds }}

как правильно макрос времени в xcom запушить?

Через templated_dict в питон операторе можно передать, в функцию аргумент(**сontext), в функции достать через контекст и запушить в xcom.

Всем привет. Может кто-нибудь подсказать, есть ли в аирфлоу возможность создать в одном операторе словарь, а затем передавать этот словарь в качестве аргументов другим операторам?

Если этот словарь сериализуемый то через xcom

Да, сериализуемый. Благодарю)

большое спасибо, помогло

Как вам airflow 2? есть смысл мигрировать?

ну рано или поздно придётся

Огонь

Быстрый шедулер

Игорь - -заинтриговал