Как мы оркестрируем процессы обработки данных с помощью Apache Airflow

[dkasyanov]

Спасибо за статью, очень вовремя!
Ваше описание того, как соотносятся execution_date и start_date, и что вообще происходит, оказалось наиболее понятным из тех, что я встречал на русском и английском. Хотя подход Airflow пока не очень хорошо в голове укладывается.

но сейчас мы перешли на подход, при котором вшиваем расписание прямо в dag_id

Как-то автоматически делается или руками?

Правильно ли понимаю, что идемпотентность достигается за счет использования {{ ds }}, как в примере про Hive?

Сейчас пробую Airflow на замену куче кронтабов. Сейчас данные перекладываются по принципу: «каждые n минут во все системы докидываются новые данные с момента последней синхронизации». Правильно ли я понимаю, что при использовании airflow больше подходит DAG/task вида «Загрузить данные за такой-то час такого-то дня (основанный, например, на {{ ts }}) » с расписанием раз в час? Час взят для примера.

Может ли второй подход сработать для интервалов в 10 минут и стоит ли его применять?

Еще вопрос про изменение {{ dag_id }}. Правильно ли я понимаю, что после того как scheduler прочитает обновленную конфигурацию DAGов, если {{ dag_id }} будет новый, то следующего запуска (и всех последующих) старой версии DAG не будет, а новая версия будет запускаться согласно расписанию. Но если оставить {{ dag_id }} старым, то будет что-то странное?

Могут ли возникнуть проблемы с расписанием, если поменять название c X на Y и потом обратно на X? Не сталкивались?

Надеюсь, что комменатрий не слишком напоминает вопрос для StackOverflow :)

[sowow359]

Как-то автоматически делается или руками?

Автоматически, на основе schedule_interval

Правильно ли понимаю, что идемпотентность достигается за счет использования {{ ds }}, как в примере про Hive?

Всё верно, за счет макросов, основанных на execution_date

Правильно ли я понимаю, что при использовании airflow больше подходит DAG/task вида «Загрузить данные за такой-то час такого-то дня (основанный, например, на {{ ts }}) » с расписанием раз в час?

Всё правильно понимаете:) Можете забирать все данные, например, между execution_date и next_execution_date.

Может ли второй подход сработать для интервалов в 10 минут и стоит ли его применять?

It depends, возможно, для вашей задачи это подойдет. Для таких интервалов запуска нужно помнить, что задачи в некоторых случаях могут не успеть отработать за это время, и пайплайн будет опаздывать, поэтому пробуйте на свой страх и риск. Если всё-таки нужно что-то около-реалтаймовое, то имеет смысл посмотреть на другие инструменты

Правильно ли я понимаю, что после того как scheduler прочитает обновленную конфигурацию DAGов, если {{ dag_id }} будет новый, то следующего запуска (и всех последующих) старой версии DAG не будет, а новая версия будет запускаться согласно расписанию.

Если вы заменили и schedule_interval, и dag_id, то у вас просто добавится новый даг, а старый будет недоступен (он останется в UI, но если попробовать в него зайти, то вылетит ошибка вида DAG seems to be missing, запускаться он соответственно тоже не будет). А вот если вы поменяли только schedule_interval, то что-то странное очень вероятно:) Применяйте версионирование и все будет работать хорошо.

Могут ли возникнуть проблемы с расписанием, если поменять название c X на Y и потом обратно на X? Не сталкивались?

При изменении названия (только названия) с X на Y даг с названием X перестанет запускаться, активным станет Y. Соответственно, при изменении обратно на X снова станет активным старый пайплайн, и будет запускаться без проблем, разве что несколько запусков могут быть пропущены.

[dkasyanov]

между execution_date и next_execution_date

Спасибо за идею! Задавая вопрос, я предполагал как-то парсить ts, но ваш вариант гораздо круче.

[vadim_vyacheslavovich]

DAG, он же пайплайн, он же направленный циклический граф

Статья хорошая, но все же маленькое замечание: Airflow все-таки оперирует ациклическими графами.

[sowow359]

Спасибо за замечание, исправились:)

[andrecpc]

Спасибо. Актуально.

[itrader]

Есть общие функции которые используются в множестве дагов. Как их можно организовать? В каком виде они могут быть оформлены в контексте Airflow?

[sowow359]

У нас для этого есть модуль libs, который лежит в репозитории (и на нодах эйрфлоу) рядом с дагами, туда мы кладем все наши генераторы, кастомные операторы и прочие штуки, которые могут как-то переиспользоваться в разных местах в коде пайплайнов