ну и с каким-нибудь селери экзекъютором

вариант с разделением- не подходит. Это конечно правильно, и про селери или кубер.
Но все же хочется полечить проблему именно в такой конфигурации сервера и экзекьютора

Она по идее должна решаться в теории

а какой экзекьютор, если про брокер ответили "да"?

Наткнулся на неприятную фичу в айрфлоу с выполнением команды vacuum table на БД Greenplum, в ситуациях, когда айрфлоу не является владельцем таблицы.

Если вакуум выполняется внутри пайтон_оператора, через cursor.execute('vacuum table'), то якобы всё проходит ок, без каких либо сообщений об ошибке. Но это похоже проблема psycopg2, так как локально проходит так же.

Если вакуум выполняет постгрес_оператор, варнинг есть, но таск всё равно помечался как выполненный
Это из логов:
[2020-06-22 08:00:59,730] {logging_mixin.py:112} INFO - [2020-06-22 08:00:59,730] {dbapi_hook.py:174} INFO - vacuum analyze sales.rid_status_log;
[2020-06-22 08:00:59,946] {postgres_operator.py:67} INFO - WARNING:  skipping "rid_status_log" --- only table or database owner can vacuum it
[2020-06-22 08:00:59,952] {taskinstance.py:1048} INFO - Marking task as SUCCESS.dag_id=v1.gp_LOAD_INC_pg_nats, task_id=vacuum_analyze_rid_status_log, execution_date=20200622T041530, start_date=20200622T050059, end_date=20200622T050059

Можно ли путём настройки конфига или параметров дага/таска сделать так, чтобы в подобных случаях таск получал статус failed?

Короче, скорее всего дело в том, что не-воркеры выдирают всё

Попробуйте ограничить в ресурсах вебсервер и шедулер

...и перейдите на селери

По-моему, эта проблема в целом касается гринплама. Можете только кодом дописать

Печалька. Спасибо - хотя я надеялся на другой ответ :)

Ну да, периодически проскальзывает вообще везде. Думаю, что только запрос на селект предварительный может как-то помочь

Добрый день! Появилась возможность в проекте попробовать airflow
Но пока не понимаю, как подобную схему сделать на airflow. Подскажите.

1. По шедулеру запускается сканировние. Из внешней системы получаем идентификаторы объектов по которым были изменения (создались новые/обновились уже загруженные). Для каждого идентификатора создается задача в очереди

2. Воркеры получают задачи из очереди, загружают соответствующие данные из внешней системы, преобразовывают, сохраняют данные в БД.

Как в такой схеме правильно использовать start_date DAG'а (или, возможно, схема должна быть принципиально другой?)
И как распараллелить обработку изменений по нескольким воркерам?

выглядит как попытка натянуть найфай на эйрфлоу

или селери
или префект

Стримсеты ещё)

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

Вы слишком много думаете за систему. Вам надо знать лишь по какому условию запускать задачу и написать код, который будет разруливать что в каком случае выполнять. Если вам нужна бОльшая детализация, то да, не стоит тратить время на аерфлоу

а что делать? писать самописные сервисы поверх стандартной очереди (redis? rabbit? kafka?)

да ну, селери справится прекрасно

там и мониторинг есть