Ну зависит от того, как у вас развёрнуто. В кубернетесе для этого есть git-sync, для celery либо добавлять в образ даги, либо маунтить каким-то образом (это уже зависит от вашей фантазии)

В смысле, в качестве ci-инструмента сам докер используется?

Gitlab CI/CD -> Docker image -> AWS ECR -> AWS ECS
для образа
Даги монтируем через AWS EFS

А, ок

И ещё такой вопрос, кто-нибудь сталкивался с проблемой, что если даги/плагины появляются в ФС позже (используется git-sync, так как все живет в k8s), чем стартует airflow webserver/scheduler, то в UI висят ошибки, якобы импортируются неизвестые плагины, хотя при этом scheduler/worker отрабатывают как надо.

Удалось решить проблему добавлением задержки перед стартом airflow-компонент, есть ещё вариант в init-контейнере это дело инициализировать, но тоже не очень нравится вариант.

Настройки: AIRFLOW__CORE__LAZY_LOAD_PLUGINS, AIRFLOW__WEBSERVER__RELOAD_ON_PLUGIN_CHANGE, AIRFLOW__CORE__EXECUTE_TASKS_NEW_PYTHON_INTERPRETER
не принесли результата(

по сути чисто UIная проблема, но глаза мазолит

Сделайте рестарт шедулера через пару секунд после запуска сервера, или отложенный запуск, если возможно

Именно шедулер читает даги

Ну вот отложенным запуском и приходится пользоваться

Ребят, такая проблема у меня. Не могу понять что за эксепшен.

Смысл в том что ровно в 00:00 по utc у меня фейляться даги.

При этом логов в самом airflow по таким таскам логов нет.

Мой обработчик ошибок ловит следующее:

Executor reports task instance <TaskInstance: exa_REPLICA.WEBAPP_REFERRAL_TRANSACTIONS_SUMS 2021-04-01 23:00:00+00:00 [queued]> finished (failed) although the task says its queued. Was the task killed externally?

Кто-то сталкивался? Куда копать? Сервера не ребутяться

Может быть дело в расписании? Наступает время когда они уже не могут запустится и все ..

Подскажите, кто нибудь запускал долгий скрипт и проверяли ли вы output ? Как это лучше сделать? Писать output в  файл и читать этот файл?

А логи?

Да как вариант. Спасибо

Всем привет. Я DS, мне посоветовали Airflow для моего кейса по предиктивному мониторингу оборудования.

Хочу сделать потоковую обработку  данных со сложной логикой фича инженеринга + инференс МЛ моделей по расписанию. Кажется это про этот инструмент, верно?

А то еще есть вариант с Kafka + Apache Beam

потоковая обработка на airflow звучит неожиданно

есть откуда взять поток? оборудование чего-то замеряет и пишет в кафку?

Поток идет батчами из киттенхауса в кликхаус, можно его дублировать и получать для реал-тайм процессинга

Да, датчики стоят на оборудовании, пишут в кликхаус все

у нас вместо кафки самописный скрипт на ПХП, который проксирует с датчиков в киттенхаус)

А можно ссылочку на контейнер? Или его тег.