airflow

0зной

1ответ

Воздушный поток - BashOperator: получение ошибки «Dag run are deadlocked for DAG: ..»

Действительно нравится планировщик рабочего процесса Airflow, но при запуске с ошибкой работает простая DAG: «{jobs.py:538} ERROR - Dag running заблокированы для DAG: TEST_SCHEDULER_DAG ". Это новая у

2зной

3ответ

Возможно ли, чтобы планировщик Airflow сначала завершил цикл предыдущего дня, прежде чем начать следующий?

Прямо сейчас узлы в моей DAG переходят к задаче следующего дня, пока остальные узлы этой DAG не заканчиваются. Есть ли способ подождать, пока остальная группа DAG закончит, прежде чем перейти к циклу

2зной

1ответ

Airflow веб-сервер не начинает только в режиме отладки

Airflow веб-сервер запустить только в режиме отладки: airflow webserver -p 8051 И traceback я получаю: [2016-12-07 03:38:48,056] {__init__.py:36} INFO - Using executor CeleryExecutor [2016-12-07 03:38

2зной

1ответ

Ошибка выполнения DAG в воздушном потоке при выполнении задачи для повторной попытки

В моих аргументах по умолчанию для DAG я задал параметры retry и retry_delay. Когда я наблюдаю за пользовательским интерфейсом, при сбое задачи он кратко меняет состояние на «повтор», но сразу же посл

1зной

2ответ

Примеры воздушных потоков без дат

Я новичок в использовании airflow. Я пытаюсь запустить dag и не хочу делать какие-либо расписания. Я хотел бы запустить конвейер с аргументом командной строки и переопределить весь текущий вывод. У ме

0зной

2ответ

Планировщик воздушного потока медленный на 1 интервал

У меня есть пара графиков, которые медленны на один интервал. Моя конфигурация выглядит как args = { 'owner' : 'test', 'start_date' : datetime.now(), 'email' : ['[email protected]'],

3зной

2ответ

Я читал часто задаваемые вопросы по Airflow о "What's the deal with start_date?", но мне все еще не ясно, почему это рекомендуется использовать динамически start_date. В моем понимании, DAG-х executio

1зной

1ответ

Запускает ли планировщик воздушного потока предыдущий запуск DAG при введении новых задач в DAG?

У меня есть определение DAG, которая динамически добавляет задачи из конфигурации в DAG, например: for k, v in config.iteritems(): bash_task = BashOperator(task_id='{}_task'.format(k), ...)