Перевести Airflow на официальный провайдер ClickHouse #124

New Issue

2026-08-21T22:49:00+03:00

ddmitry commented

2026-08-21 22:49:00 +03:00

Следует за #119. Жизненный цикл мира уже принят целиком; эта задача меняет
одну границу — способ, которым даги исполняют SQL ClickHouse.

Цель

Перевести все исполняемые SQL-файлы ClickHouse, которыми пользуются даги, на
официальный провайдер Airflow apache-airflow-providers-clickhousedb. Файлы
напрямую получает штатный SQLExecuteQueryOperator: даг задаёт граф, порядок и
параметры, а преобразование остаётся в самодостаточном SQL своего слоя.
Программная логика с ветвлением, ожиданием или разбором результата использует
официальный ClickHouseHook.

Менти должен видеть один штатный путь SQL-интеграции Airflow, а не разбираться
в проектном разборщике, ручном открытии клиента или скрытой сборке одного
запроса из нескольких файлов.

Официальный провайдер уже совместим со стендом: Airflow 3.3.0,
apache-airflow-providers-common-sql 2.0.1 и clickhouse-connect 1.6.0
удовлетворяют его требованиям. На момент постановки официальный реестр
содержит версии 1.0.0 и 1.0.1.

Сначала прочитать

docs/specs/2026-08-21-world-lifecycle.md;
docs/adr/0013-world-lifecycle.md;
docs/adr/0001-stand-services.md и docs/adr/0007-clickhouse-access.md;
docs/architecture/storage.md, раздел «Раскладка SQL»;
docs/architecture/orders/ingestion.md;
dags/world_lifecycle.py, dags/orders_ingest.py и
dags/test_clickhouse.py;
sql/ddl/*.sql, sql/stg/orders_raw_load.sql,
sql/ods/order_load.sql, sql/ods/order_errors_load.sql и
sql/ods/_order_wire_contract.sql;
infra/airflow/Dockerfile и infra/airflow/init.sh;
официальную документацию apache-airflow-providers-clickhousedb,
ClickHouseHook и SQLExecuteQueryOperator.

Критерии приёмки

В образе Airflow закреплена совместимая версия
apache-airflow-providers-clickhousedb. Драйвер
clickhouse-connect==1.6.0 остаётся прямым пином: провайдер задаёт только
нижнюю границу версии, а сборка стенда должна быть воспроизводимой.
pip check проходит, а итоговые версии Airflow, провайдера,
common-sql и clickhouse-connect названы в отчёте проверки.
Подключения clickhouse_default и clickhouse_lifecycle создаются с
типом clickhouse, который регистрирует провайдер.
Все исполняемые SQL-файлы нынешних дагов передаются непосредственно в
SQLExecuteQueryOperator; Python-задачи не получают отрендеренный SQL
аргументом и не исполняют его вручную.
Канонические файлы sql/ddl/*.sql исполняются с
split_statements=True в порядке имён файлов. Порядок виден в графе
world_initialize: каждый файл представлен отдельной последовательно
связанной задачей.
orders_ingest исполняет sql/stg/orders_raw_load.sql штатным
оператором и передаёт run_id как параметр ClickHouse load_id.
Две ветви STG → ODS собраны в одном самодостаточном исполняемом файле
sql/ods/order_parse.sql: сначала он пишет годные строки, затем брак. Общий
предикат объявлен локальной Jinja-макрокомандой в этом же файле и
разворачивается в оба выражения; отдельного _order_wire_contract.sql
больше нет.
Разбор ODS остаётся одной задачей parse_batch, исполняет оба выражения
последовательно с split_statements=True и передаёт им один load_id.
Самодельная функция _ddl_statements удалена. В dags/ нет прямых
импортов clickhouse_connect и вызовов clickhouse_connect.get_client().
Программные запросы, которым нужен результат, цикл ожидания или
специальная возможность ClickHouse, идут через официальный
ClickHouseHook.
world_initialize по-прежнему применяет весь канонический DDL от имени
lifecycle; частичный отказ не переводит позицию мира в готовое состояние.
world_recreate по-прежнему удаляет только прикладные базы и сохраняет
контракт позиций -1 → 0 → WORLD_STARTING_DAYS.
Документация закрепляет общую конвенцию: исполняемый SQL по умолчанию
самодостаточен и подключается оператором, а Python остаётся оркестрацией.
Обновлены как минимум docs/specs/2026-08-21-world-lifecycle.md, ADR 0001,
docs/architecture/storage.md, docs/architecture/orders/ingestion.md и
комментарии образа Airflow; остальные упоминания найдены поиском и сверены.

Границы

Содержание и раскладку файлов DDL не менять.
Смысл запросов приёма заказов, два шага pull_batch → parse_batch,
классификацию годных строк и брака и контракт _load_id не менять.
Состояния жизненного цикла, порядок очистки Kafka, генератор и цели make
не менять.
Новый оператор или обёртку над официальным провайдером не писать.
Постоянное представление или таблицу-классификатор между STG и ODS ради
общего предиката не заводить.
Общий мигратор схемы не заводить: на готовом мире make up по-прежнему не
применяет изменённый DDL.
make clean не запускать: провайдер не меняет постоянное состояние служб,
а проверка укладывается в make rebuild-storage.

Подводные камни

Перед реализацией снова сверить версию и API через MCP Context7:
resolve-library-id → query-docs, затем с официальной документацией
Airflow. Коротко записать выбранную версию и причину.
SQLExecuteQueryOperator принимает путь к шаблону относительно каталога
поиска DAG. Смонтированный sql/ должен быть доступен всем компонентам,
которые разбирают и исполняют задачу.
Airflow сначала разворачивает Jinja, затем провайдер разбивает готовый SQL на
выражения. Макрокоманда разбора заказов живёт внутри одного файла; параметр
ClickHouse {load_id:String} передаётся через parameters, а не подставляется
строкой через Jinja.
split_statements=True исполняет выражения последовательно, без общей
транзакции: ClickHouse не поддерживает многооперационные транзакции. Это
сохраняет нынешнее поведение, но не обещает откат уже выполненных выражений.
При частичном отказе Airflow повторяет весь parse_batch с тем же load_id.
Обе ветви STG → ODS остаются одним parse_batch: это одна классификация
неизменного среза. Два task_id не добавили бы транзакционности, зато снова
потребовали бы дублировать предикат, собирать запрос из нескольких файлов
либо заводить постоянный классификатор.
Провайдер свежий. Установка не должна молча обновить Airflow или соседние
провайдеры; итог разрешения зависимостей проверяется в собранном образе.

Проверка

make config-test
make lint
docker compose build airflow-init
docker compose up --detach --wait --wait-timeout 600
docker compose exec -T airflow-scheduler airflow dags list-import-errors --output json
make rebuild-storage
make smoke
make check-clickhouse
make check-services

На живом стенде дополнительно проверить:

В графе world_initialize видны последовательные задачи для всех файлов
sql/ddl/.
В графе orders_ingest остались два шага pull_batch → parse_batch, а
отрендерированный SQL второго шага содержит обе ветви и не содержит
неразвёрнутых Jinja-конструкций.
В журнале ClickHouse DDL выполнен пользователем lifecycle, а запросы
orders_ingest — пользователем подключения clickhouse_default.
Повторный make up не применяет DDL и не меняет данные.
Многозапросные файлы исполняются целиком: после пересборки существуют
объекты из начала и конца хотя бы одного DDL-файла, а управляемая порция
заказов раскладывается и в годную ветвь, и в брак.

Следует за #119. Жизненный цикл мира уже принят целиком; эта задача меняет одну границу — способ, которым даги исполняют SQL ClickHouse. ## Цель Перевести все исполняемые SQL-файлы ClickHouse, которыми пользуются даги, на официальный провайдер Airflow `apache-airflow-providers-clickhousedb`. Файлы напрямую получает штатный `SQLExecuteQueryOperator`: даг задаёт граф, порядок и параметры, а преобразование остаётся в самодостаточном SQL своего слоя. Программная логика с ветвлением, ожиданием или разбором результата использует официальный `ClickHouseHook`. Менти должен видеть один штатный путь SQL-интеграции Airflow, а не разбираться в проектном разборщике, ручном открытии клиента или скрытой сборке одного запроса из нескольких файлов. Официальный провайдер уже совместим со стендом: Airflow 3.3.0, `apache-airflow-providers-common-sql` 2.0.1 и `clickhouse-connect` 1.6.0 удовлетворяют его требованиям. На момент постановки официальный реестр содержит версии 1.0.0 и 1.0.1. ## Сначала прочитать - `docs/specs/2026-08-21-world-lifecycle.md`; - `docs/adr/0013-world-lifecycle.md`; - `docs/adr/0001-stand-services.md` и `docs/adr/0007-clickhouse-access.md`; - `docs/architecture/storage.md`, раздел «Раскладка SQL»; - `docs/architecture/orders/ingestion.md`; - `dags/world_lifecycle.py`, `dags/orders_ingest.py` и `dags/test_clickhouse.py`; - `sql/ddl/*.sql`, `sql/stg/orders_raw_load.sql`, `sql/ods/order_load.sql`, `sql/ods/order_errors_load.sql` и `sql/ods/_order_wire_contract.sql`; - `infra/airflow/Dockerfile` и `infra/airflow/init.sh`; - официальную документацию `apache-airflow-providers-clickhousedb`, `ClickHouseHook` и `SQLExecuteQueryOperator`. ## Критерии приёмки - [x] В образе Airflow закреплена совместимая версия `apache-airflow-providers-clickhousedb`. Драйвер `clickhouse-connect==1.6.0` остаётся прямым пином: провайдер задаёт только нижнюю границу версии, а сборка стенда должна быть воспроизводимой. - [x] `pip check` проходит, а итоговые версии Airflow, провайдера, `common-sql` и `clickhouse-connect` названы в отчёте проверки. - [x] Подключения `clickhouse_default` и `clickhouse_lifecycle` создаются с типом `clickhouse`, который регистрирует провайдер. - [x] Все исполняемые SQL-файлы нынешних дагов передаются непосредственно в `SQLExecuteQueryOperator`; Python-задачи не получают отрендеренный SQL аргументом и не исполняют его вручную. - [x] Канонические файлы `sql/ddl/*.sql` исполняются с `split_statements=True` в порядке имён файлов. Порядок виден в графе `world_initialize`: каждый файл представлен отдельной последовательно связанной задачей. - [x] `orders_ingest` исполняет `sql/stg/orders_raw_load.sql` штатным оператором и передаёт `run_id` как параметр ClickHouse `load_id`. - [x] Две ветви STG → ODS собраны в одном самодостаточном исполняемом файле `sql/ods/order_parse.sql`: сначала он пишет годные строки, затем брак. Общий предикат объявлен локальной Jinja-макрокомандой в этом же файле и разворачивается в оба выражения; отдельного `_order_wire_contract.sql` больше нет. - [x] Разбор ODS остаётся одной задачей `parse_batch`, исполняет оба выражения последовательно с `split_statements=True` и передаёт им один `load_id`. - [x] Самодельная функция `_ddl_statements` удалена. В `dags/` нет прямых импортов `clickhouse_connect` и вызовов `clickhouse_connect.get_client()`. - [x] Программные запросы, которым нужен результат, цикл ожидания или специальная возможность ClickHouse, идут через официальный `ClickHouseHook`. - [x] `world_initialize` по-прежнему применяет весь канонический DDL от имени `lifecycle`; частичный отказ не переводит позицию мира в готовое состояние. - [x] `world_recreate` по-прежнему удаляет только прикладные базы и сохраняет контракт позиций `-1` → `0` → `WORLD_STARTING_DAYS`. - [x] Документация закрепляет общую конвенцию: исполняемый SQL по умолчанию самодостаточен и подключается оператором, а Python остаётся оркестрацией. Обновлены как минимум `docs/specs/2026-08-21-world-lifecycle.md`, ADR 0001, `docs/architecture/storage.md`, `docs/architecture/orders/ingestion.md` и комментарии образа Airflow; остальные упоминания найдены поиском и сверены. ## Границы - Содержание и раскладку файлов DDL не менять. - Смысл запросов приёма заказов, два шага `pull_batch` → `parse_batch`, классификацию годных строк и брака и контракт `_load_id` не менять. - Состояния жизненного цикла, порядок очистки Kafka, генератор и цели `make` не менять. - Новый оператор или обёртку над официальным провайдером не писать. - Постоянное представление или таблицу-классификатор между STG и ODS ради общего предиката не заводить. - Общий мигратор схемы не заводить: на готовом мире `make up` по-прежнему не применяет изменённый DDL. - `make clean` не запускать: провайдер не меняет постоянное состояние служб, а проверка укладывается в `make rebuild-storage`. ## Подводные камни - Перед реализацией снова сверить версию и API через MCP Context7: `resolve-library-id` → `query-docs`, затем с официальной документацией Airflow. Коротко записать выбранную версию и причину. - `SQLExecuteQueryOperator` принимает путь к шаблону относительно каталога поиска DAG. Смонтированный `sql/` должен быть доступен всем компонентам, которые разбирают и исполняют задачу. - Airflow сначала разворачивает Jinja, затем провайдер разбивает готовый SQL на выражения. Макрокоманда разбора заказов живёт внутри одного файла; параметр ClickHouse `{load_id:String}` передаётся через `parameters`, а не подставляется строкой через Jinja. - `split_statements=True` исполняет выражения последовательно, без общей транзакции: ClickHouse не поддерживает многооперационные транзакции. Это сохраняет нынешнее поведение, но не обещает откат уже выполненных выражений. При частичном отказе Airflow повторяет весь `parse_batch` с тем же `load_id`. - Обе ветви STG → ODS остаются одним `parse_batch`: это одна классификация неизменного среза. Два `task_id` не добавили бы транзакционности, зато снова потребовали бы дублировать предикат, собирать запрос из нескольких файлов либо заводить постоянный классификатор. - Провайдер свежий. Установка не должна молча обновить Airflow или соседние провайдеры; итог разрешения зависимостей проверяется в собранном образе. ## Проверка ```text make config-test make lint docker compose build airflow-init docker compose up --detach --wait --wait-timeout 600 docker compose exec -T airflow-scheduler airflow dags list-import-errors --output json make rebuild-storage make smoke make check-clickhouse make check-services ``` На живом стенде дополнительно проверить: 1. В графе `world_initialize` видны последовательные задачи для всех файлов `sql/ddl/`. 2. В графе `orders_ingest` остались два шага `pull_batch` → `parse_batch`, а отрендерированный SQL второго шага содержит обе ветви и не содержит неразвёрнутых Jinja-конструкций. 3. В журнале ClickHouse DDL выполнен пользователем `lifecycle`, а запросы `orders_ingest` — пользователем подключения `clickhouse_default`. 4. Повторный `make up` не применяет DDL и не меняет данные. 5. Многозапросные файлы исполняются целиком: после пересборки существуют объекты из начала и конца хотя бы одного DDL-файла, а управляемая порция заказов раскладывается и в годную ветвь, и в брак.

ddmitry added the ready-for-agent label 2026-08-21 22:49:00 +03:00

ddmitry added a new dependency 2026-08-21 22:49:10 +03:00

#119 Отдать жизненный цикл мира дагам Airflow

ddmitry referenced this issue

2026-08-21 23:27:59 +03:00

Передать жизненный цикл мира Airflow #125

ddmitry self-assigned this 2026-08-22 00:28:59 +03:00

ddmitry referenced a pull request that will close this issue

2026-08-22 09:21:18 +03:00

Подключить официальный провайдер ClickHouse к Airflow #126

ddmitry closed this issue

2026-08-22 10:06:07 +03:00