E
airflow initdb делали?
у меня без этого новые даги не добавляются
у меня без этого новые даги не добавляются
к сожалению, не помогло
Size: a a a
2020 July 29
ИХ
к сожалению, не помогло
через докер развернуто?
E
через докер развернуто?
угу
PA
Вот он висит и ждёт, когда прошлый запуск завершится))) А его вообще нет)
дага с экстернал сенсором стартует одновременно с той дагой которую она прослушивает
ИХ
подскажите, как правильно импортировать свои пакеты?
разложил пока даги в такую структуру:
|—dags
| |—pr_1
| |—__init__.py
| |—dag1.py
| |—my_module.py
|—__init__.py
пишу в dag1 - from dags.pr_1.my_module import MyClass
а airflow мне - "No module named 'dags.pr_1'"
пока обычный пукелевский контейнер, добавил PYTHONPATH только в Dockerfile
ENV PYTHONPATH=${AIRFLOW_HOME}:${PYTHONPATH}
разложил пока даги в такую структуру:
|—dags
| |—pr_1
| |—__init__.py
| |—dag1.py
| |—my_module.py
|—__init__.py
пишу в dag1 - from dags.pr_1.my_module import MyClass
а airflow мне - "No module named 'dags.pr_1'"
пока обычный пукелевский контейнер, добавил PYTHONPATH только в Dockerfile
ENV PYTHONPATH=${AIRFLOW_HOME}:${PYTHONPATH}
читайте вниз
ME
дага с экстернал сенсором стартует одновременно с той дагой которую она прослушивает
Там в примере она сама себя слушает)
ME
Наверное надо свой сенсор написать, который обработает кейс первого запуска и будет слушать предыдущий результат
PA
Наверное надо свой сенсор написать, который обработает кейс первого запуска и будет слушать предыдущий результат
стоп, нужно слушать внутри дага или разные даги?
ME
стоп, нужно слушать внутри дага или разные даги?
Внутри одного дага предыдущий запуск. Чтобы в параллели не работало несколько
PA
то есть если прошлый запуск дага был успешным то выполнять по расписанию завтра например так?
SK
тестовый пример взлител, там походу только первый запуск нужно отметить как успешный(для сенсора)
ME
то есть если прошлый запуск дага был успешным то выполнять по расписанию завтра например так?
Ага, но скорее для бэкфила актуально
ME
А если прошлый зафейлился, то ничего не делать
2020 July 30
A
Подскажите пожалуйста, реализовывал ли кто-то механизм восстановления после падения сервера с AirFlow?
Кейс: есть задача (например запись данных в Greenplum), выполняемая несколько часов, во время выполнения задачи AirFlow падает. Восстанавливаем (если в k8s, то довольно быстро, но всё равно не моментально + это уже "другой" AirFlow в части подключения к Greenplum), и не можем получить никакой информации по статусу выполнения нашей задачи, а за время недоступности она могла быть: a) выполнена, b) ещё находится в процессе выполнения, c) отвалиться.
Понятно, что очевидным выходом из ситуации будет перезапуск всех задач, но на это уйдёт много времени и хотелось бы как-то восстановить состояние до падения.
Может ли как-то помочь DAG Serialization?
Кейс: есть задача (например запись данных в Greenplum), выполняемая несколько часов, во время выполнения задачи AirFlow падает. Восстанавливаем (если в k8s, то довольно быстро, но всё равно не моментально + это уже "другой" AirFlow в части подключения к Greenplum), и не можем получить никакой информации по статусу выполнения нашей задачи, а за время недоступности она могла быть: a) выполнена, b) ещё находится в процессе выполнения, c) отвалиться.
Понятно, что очевидным выходом из ситуации будет перезапуск всех задач, но на это уйдёт много времени и хотелось бы как-то восстановить состояние до падения.
Может ли как-то помочь DAG Serialization?
VS
логически сериализации не поможет никак
VS
опять таки, что именно упало у вас в airflow? все таки там не один сервис
A
опять таки, что именно упало у вас в airflow? все таки там не один сервис
все сервисы на 1 машине (и в ближайшее время нет возможности их разнести), соответственно падают все разом.
VS
то у вас кубик, то монолит на одной машине 🤔
A
кубик - состояние to-be)
A
может я некорректно написал, допустим мы перевели AirFlow на k8s, всё равно вопрос на мой взгляд остаётся открытым.