Всем привет. Подскажите ответы на три вопроса, хотя бы ссылками на доки или Ютуб:
1. Можно ли как то интерактивно взаимодействовать с запущенным вручную дагом кроме как переменными из админки? К примеру указать имя БД и таблицы для текущего выполнения?
2. Я где то читал, что есть готовый оператор или хук для запуска задачи в Talent ETL из airflow, но теперь не могу найти, есть ли тут кто то, кто в курсе?
3. Возможно вопрос не сюда, но каким образом лучше переливать данные из mongo в реляционную БД, при условии, что один файл в mongo около 10 Гб может быть, а перелить надо несколько таких файлов? Придется где то приземлять около airflow в промежуточную базу? На практике пробовал через pandas, но на 10-м файле может упасть так как очистка памяти в питоне как то не полностью все вычищает после закрытия файла и уничтожения датафрейма. gc.collect тоже не помогает.

Спасибо

Имел ввиду чтобы юзер не знающий питон, мог дернуть его руками выбрав в раскрывающимся списке какой то параметр перед запуском.

Всем привет. Подскажите ответы на три вопроса, хотя бы ссылками на доки или Ютуб:
1. Можно ли как то интерактивно взаимодействовать с запущенным вручную дагом кроме как переменными из админки? К примеру указать имя БД и таблицы для текущего выполнения?
2. Я где то читал, что есть готовый оператор или хук для запуска задачи в Talent ETL из airflow, но теперь не могу найти, есть ли тут кто то, кто в курсе?
3. Возможно вопрос не сюда, но каким образом лучше переливать данные из mongo в реляционную БД, при условии, что один файл в mongo около 10 Гб может быть, а перелить надо несколько таких файлов? Придется где то приземлять около airflow в промежуточную базу? На практике пробовал через pandas, но на 10-м файле может упасть так как очистка памяти в питоне как то не полностью все вычищает после закрытия файла и уничтожения датафрейма. gc.collect тоже не помогает.

Спасибо

2. Возможно это

https://www.astronomer.io/guides/airflow-talend-integration

Добрый вечер всем!
Генерируем динамический dag. Сам .py файл с dag'ом лезет в базу за параметрами тасков. Как ограничить частоту с которой Airflow 1.10.10 запускает (перечитывает) этот .py файл? Получается, что он обращается к базе несколько раз в секунду.

Выставил

# The number of seconds to wait between consecutive DAG file processing
processor_poll_interval = 30

# after how much time (seconds) a new DAGs should be picked up from the filesystem
min_file_process_interval = 30

# How often (in seconds) to scan the DAGs directory for new files. Default to 5 minutes.
dag_dir_list_interval = 30

Не помогает. Airflow рестартовал, естественно. Так и спамит базу запросами.

Всем привет. Подскажите ответы на три вопроса, хотя бы ссылками на доки или Ютуб:
1. Можно ли как то интерактивно взаимодействовать с запущенным вручную дагом кроме как переменными из админки? К примеру указать имя БД и таблицы для текущего выполнения?
2. Я где то читал, что есть готовый оператор или хук для запуска задачи в Talent ETL из airflow, но теперь не могу найти, есть ли тут кто то, кто в курсе?
3. Возможно вопрос не сюда, но каким образом лучше переливать данные из mongo в реляционную БД, при условии, что один файл в mongo около 10 Гб может быть, а перелить надо несколько таких файлов? Придется где то приземлять около airflow в промежуточную базу? На практике пробовал через pandas, но на 10-м файле может упасть так как очистка памяти в питоне как то не полностью все вычищает после закрытия файла и уничтожения датафрейма. gc.collect тоже не помогает.

Спасибо

А он на воркерах не создаёт ещё раз даг при работе?

3. Используйте средство для закачивания кучи данных файлом именно этой базы. У каждой есть такой. Например, для постгри - copy,   для гринплама gpfdist, ну и в таком роде

Спасибо за направление поиска. Еще вопрос тогда: а можно ли генерить задачи в даге "на лету"? Попробую пояснить свою идею: например первая задача-сенсор в даге смотрит в папку и видит там 5 новых файлов. Потом создается 5 новых последовательных задач с python env operator на каждый файл для дальнейшей их обработки. Мне кажется таким образом у нас лучше соблюдается условие идемпотентности и в случае чего легче откатиться если упадёт задача работавшаятс одним файлом. Если путаюсь в терминологии то сори, только начал изучать этот инструмент.

Спасибо за направление поиска. Еще вопрос тогда: а можно ли генерить задачи в даге "на лету"? Попробую пояснить свою идею: например первая задача-сенсор в даге смотрит в папку и видит там 5 новых файлов. Потом создается 5 новых последовательных задач с python env operator на каждый файл для дальнейшей их обработки. Мне кажется таким образом у нас лучше соблюдается условие идемпотентности и в случае чего легче откатиться если упадёт задача работавшаятс одним файлом. Если путаюсь в терминологии то сори, только начал изучать этот инструмент.

Но мне кажется, что вы не там ищете решение. При репликации данных аерфлоу нужен только для передачи файлов с помощью ssh \ sftp \ hdfs \ etc с сервера монги на целевой, а потом отправку запроса, упакованного в одну транзакцию, который вычитывает в базу все файлы

Добрый вечер всем!
Генерируем динамический dag. Сам .py файл с dag'ом лезет в базу за параметрами тасков. Как ограничить частоту с которой Airflow 1.10.10 запускает (перечитывает) этот .py файл? Получается, что он обращается к базе несколько раз в секунду.

Выставил

# The number of seconds to wait between consecutive DAG file processing
processor_poll_interval = 30

# after how much time (seconds) a new DAGs should be picked up from the filesystem
min_file_process_interval = 30

# How often (in seconds) to scan the DAGs directory for new files. Default to 5 minutes.
dag_dir_list_interval = 30

Не помогает. Airflow рестартовал, естественно. Так и спамит базу запросами.

Встречал такую инфу:
The Airflow executor executes top level code on every heartbeat, so a small amount of top level code can cause performance issues. Try to treat the DAG file like a config file and leave all the heavy lifting for the hook and operator.