Коллеги, привет. Есть кто может поделиться опытом использования airflow в сервинге ml моделей?

Хай. Я понимаю, что вопрос тупой и я вроде его задавал. Но я решил вернуться к Airflow и не пойму как мне отслеживать запуски дагов.

Вот у меня в системе появляются объекты. В день хз сколько штук в неопределенное вермя. Мне было бы удобно при их создании через api триггерить запуск дагов для обработки этих объектов. Но как я понимаю, единственный параметр рана - дата.

Как правильно организовать обработку объектов?

Доброго дня всем, вопрос по настройке отправки уведомлений из Airflow при превышение времени  работы daga.

Какие стандартные механизмы можно применить, и как настроить отправку сообщения ? ( отдельным таском)

Триггерить по апи можно, передавая в конфиге даграна информацию об объектах. http://airflow.apache.org/docs/apache-airflow/1.10.2/api.html

Но это плохая практика: если будет спайк в нагрузке, то даг раны просто встанут (испытано на собственной шкуре)

sla_miss_callback в помощь

https://github.com/DFStoneburner/airflow-sla-examples

А что делать? Брать не эйрфлоу? Я хочу сразу при появлении объектов начинать их обработку не дожидаясь завершения обработки прошлого. При это параллелизовать что можно

Спасибо

я бы сказал, что путь эйрфлоу — сделать сенсоры на эти объекты (например, которые будут дергать внешний api и проходить/валиться по статус-коду ответа, мануалы по написанию сенсоров есть в оф.доке) и поставить их перед обработкой, соответственно скипая всю ветку если объекта нет. Поставить по расписанию на комфортный для вас интервал (зависит от того как часто появляются объекты и какую задержку вы готовы переждать). отслеживать запуски дагов не нужно, если вы готовы к асинхронной обработке.
Если ваше основное и необходимое требование — триггерить обработку сразу при получении объекта, то эйрфлоу вам тут не поможет, так как он не рассчитан на подобное, для шедулера допустимы паузы и сложно настроить хороший порядок исполнения разных дагов, так что проще будет сразу взять не эйрфлоу.

Ну это не совсем задача для Airflow, да. Там всё-таки расписание всему голова.
Для себя мы решили, что нам хватит и запускать раз в полчаса даг, в котором обнаруживать новые объекты и обрабатывать их.

А если говорить о параллелизации такого подхода – первый оператор – ShortCircuitOperator (ничего не нашли – не триггерим обработку), в нём раскладываем объекты в Variables вида <dag_name>_<i> (держим где-то 20 тасков, обрабатывающих всё это дело), это позволяет не иметь проблем со структурой дага (эирфлоу не любит, когда динамически меняют даг)

в смысле, заранее создать 20 копий тасков?

Можно создать циклом под объекты

Ага, там однотипные действия по обработке новых объектов, сами задачки тяжёлые, один воркер будет либо долго кряхтеть, либо вылетит по памяти

Ну циклом или не циклом - не важно. Вопрос про граф скорее

можно сразу вопрос задать пользуясь случаем?

То есть аэйрфлоу из коробки ничегь не предлагает для параллелизации?

Можно, на будущее лучше сразу сам вопрос задавать, а не метавопрос)

Ну почему не предлагает? Celery раскидывает задачки по воркерам для выполнения – вот она и параллелизация)

у меня есть неопеделенное количество объектов например. я написал пайтон функцию которая клиентом ходит на бигквери и ранает там джобы и я вот так паралелюсь.
   transfer_tasks = []
   for table in tables_list:
       transfer_tasks.append(PythonOperator(
           task_id='transfer_tasks_' + str(table),
           python_callable=data_transfer,
           op_kwargs={'table': table},
           dag=dag,
       )
       )
это норм подход?

или можно как то изящнее решить?