Воздушный поток выполняет один и тот же оператор дважды

Question

Я использую Airflow 1.8.0 с LocalExecutor, и у меня есть ежедневная группа доступности баз данных, выполняющая различные запросы Redshift с использованием postgres_operator для вставки некоторых строк в таблицу. Я только что заметил, что иногда для данного TaskInstance postgres_operator запускается дважды, поэтому в итоге в моей таблице появляются дубликаты.

Когда я просматриваю логи TaskInstance, я вижу что-то подобное для проблемных taskInstances: «Первая попытка 1 из 2» происходит дважды в одну и ту же секунду (без автоматической повторной попытки или вмешательства вручную):

Вы знаете, откуда может возникнуть проблема?

Спасибо!

Answer 1

По логам похоже, что это происходит при запуске, поэтому задание может быть заполнено.

Как пример, рекомендуется создавать идемпотентные группы обеспечения доступности баз данных, используя execution_date в своей логике. Поток воздуха может запускать их дважды в попытке «догнать». Из раздела Backfill and Catchup в документации по воздушному потоку:

Ключевой возможностью Airflow является то, что эти прогоны DAG являются атомарными, идемпотентные элементы, и планировщик по умолчанию изучит время жизни DAG (от начала до конца / сейчас, один интервал за раз) и запустить прогон DAG для любого интервала, который не был выполнен (или имеет был очищен). Эта концепция называется Catchup.

Когда мы начали, у нас была похожая проблема.

Answer 2

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