feat(airflow): подключён официальный провайдер ClickHouse

- Зачем:
  - нужен единый штатный путь исполнения SQL и программных запросов ClickHouse из Airflow.
- Что:
  - добавлен официальный провайдер и подключения типа `clickhouse`.
  - SQL-файлы переданы `SQLExecuteQueryOperator`, а программные запросы — `ClickHouseHook`.
  - разбор заказов собран в одном файле, DDL разделён по задачам, документация и смоук обновлены.
- Проверка:
  - выполнены `make lint` и `make config-test`.
  - собран образ, выполнены `pip check`, `make rebuild-storage`, `make smoke` (21/21), `make check-clickhouse` (10/10) и `make check-services` (7/7).

dags/orders_ingest.py

@@ -18,9 +18,9 @@
 from __future__ import annotations
 
 import datetime
-import logging
 
-from airflow.sdk import Connection, dag, get_current_context, task
+from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator
+from airflow.sdk import dag
 
 START_DATE = datetime.datetime(2026, 1, 1, tzinfo=datetime.UTC)
 
@@ -43,67 +43,25 @@ SQL_ROOT = "/opt/airflow/sql"
 )
 def orders_ingest():
     """Забрать приехавший слепок заказов и разложить его по слоям."""
-
-    def clickhouse_client():
-        # Импорт при выполнении задачи, а не при разборе файла: обработчик DAG
-        # разбирает его снова и снова, и импорт наверху оплачивался бы каждым
-        # разбором.
-        import clickhouse_connect
-
-        connection = Connection.get("clickhouse_default")
-        return clickhouse_connect.get_client(
-            host=connection.host,
-            port=connection.port,
-            username=connection.login,
-            password=connection.password,
-            database=connection.schema or "default",
-            connect_timeout=5,
-            send_receive_timeout=30,
-        )
-
-    # Аргумент с расширением из templates_exts Airflow подменяет текстом файла:
-    # берёт его из template_searchpath и прогоняет через Jinja при исполнении
-    # задачи, а не при разборе дага. В задачу приезжает готовый запрос — вместе
-    # с тем, что файл подключил через include.
-    @task(templates_exts=(".sql",))
-    def pull_batch(sql: str) -> None:
-        load_id = get_current_context()["run_id"]
-        client = clickhouse_client()
-        try:
-            summary = client.command(sql, parameters={"load_id": load_id})
-        finally:
-            client.close()
-        # Размер порции — read_rows: written_rows у вставки в Distributed
-        # считает не приехавшее.
-        logging.info(
-            "порция принята: строк %s, _load_id %s",
-            summary.summary["read_rows"],
-            load_id,
-        )
-
-    @task(templates_exts=(".sql",))
-    def parse_batch(good_rows_sql: str, bad_rows_sql: str) -> None:
-        """Разобрать срез сырья в версии заказов и в брак.
-
-        Обе вставки в одном task_id: транзакции между ними ClickHouse не даёт,
-        а повтор задачи безопасен — срез читается по тому же неизменному
-        _load_id (docs/architecture/orders/ingestion.md, «Поток данных»).
-        """
-        load_id = get_current_context()["run_id"]
-        client = clickhouse_client()
-        try:
-            client.command(good_rows_sql, parameters={"load_id": load_id})
-            client.command(bad_rows_sql, parameters={"load_id": load_id})
-        finally:
-            client.close()
-        # Счётчиков строк нет: у запроса с WHERE read_rows считает прочитанное
-        # с диска, а не подошедшее (storage.md, «Что проверено»).
-        logging.info("срез разобран: _load_id %s", load_id)
-
-    pull_batch("stg/orders_raw_load.sql") >> parse_batch(
-        "ods/order_load.sql",
-        "ods/order_errors_load.sql",
+    pull_batch = SQLExecuteQueryOperator(
+        task_id="pull_batch",
+        conn_id="clickhouse_default",
+        sql="stg/orders_raw_load.sql",
+        parameters={"load_id": "{{ run_id }}"},
+        do_xcom_push=False,
     )
+    parse_batch = SQLExecuteQueryOperator(
+        task_id="parse_batch",
+        conn_id="clickhouse_default",
+        sql="ods/order_parse.sql",
+        parameters={"load_id": "{{ run_id }}"},
+        split_statements=True,
+        do_xcom_push=False,
+    )
+
+    # ClickHouse не даёт транзакции между двумя вставками parse_batch. Один
+    # task_id сохраняет их порядок и одну точку повтора для неизменного load_id.
+    pull_batch >> parse_batch
 
 
 orders_ingest()

dags/test_clickhouse.py

@@ -19,6 +19,7 @@ from __future__ import annotations
 import datetime
 import uuid
 
+from airflow.providers.clickhousedb.hooks.clickhouse import ClickHouseHook
 from airflow.sdk import Connection, dag, get_current_context, task
 
 CLUSTER = "clickstream_cluster"
@@ -54,27 +55,12 @@ def _remote_parameters() -> dict[str, str]:
     }
 
 
-def _clickhouse_client():
-    # clickhouse_connect импортируется внутри функции, а не наверху файла:
-    # обработчик DAG разбирает этот файл снова и снова, и импорт наверху
-    # оплачивался бы каждым разбором. Тяжёлые импорты Airflow советует
-    # держать внутри задач.
-    import clickhouse_connect
-
-    connection = Connection.get("clickhouse_default")
-    return clickhouse_connect.get_client(
-        host=connection.host,
-        port=connection.port,
-        username=connection.login,
-        password=connection.password,
-        database=connection.schema or "default",
-        connect_timeout=5,
-        send_receive_timeout=30,
-    )
+def _clickhouse_hook() -> ClickHouseHook:
+    return ClickHouseHook(clickhouse_conn_id="clickhouse_default")
 
 
-def _table_engines(client, source: str) -> list[tuple[str, str]]:
-    result = client.query(
+def _table_engines(hook: ClickHouseHook, source: str) -> list[tuple[str, str]]:
+    return hook.get_records(
         f"""
         SELECT name, engine
         FROM {source}
@@ -84,23 +70,23 @@ def _table_engines(client, source: str) -> list[tuple[str, str]]:
         """,
         parameters=_remote_parameters(),
     )
-    return result.result_rows
 
 
-def _drop_tables(client) -> None:
-    client.command(
-        f"DROP TABLE IF EXISTS default.{DISTRIBUTED_TABLE} ON CLUSTER {CLUSTER} SYNC"
-    )
-    client.command(
-        f"DROP TABLE IF EXISTS default.{LOCAL_TABLE} ON CLUSTER {CLUSTER} SYNC"
+def _drop_tables(hook: ClickHouseHook) -> None:
+    hook.run(
+        [
+            f"DROP TABLE IF EXISTS default.{DISTRIBUTED_TABLE} "
+            f"ON CLUSTER {CLUSTER} SYNC",
+            f"DROP TABLE IF EXISTS default.{LOCAL_TABLE} ON CLUSTER {CLUSTER} SYNC",
+        ],
     )
 
 
 # Отсутствие таблиц проверяют обе задачи — перед созданием и после уборки,
 # — поэтому у этой проверки своё имя, а остальные живут прямо в теле задач.
-def _assert_tables_absent(client) -> None:
+def _assert_tables_absent(hook: ClickHouseHook) -> None:
     for node_name, source in NODES:
-        remaining = _table_engines(client, source)
+        remaining = _table_engines(hook, source)
         if remaining:
             raise RuntimeError(
                 f"служебные таблицы остались на {node_name}: {remaining}"
@@ -118,11 +104,11 @@ def _assert_tables_absent(client) -> None:
 def test_clickhouse():
     @task
     def prepare_tables() -> None:
-        client = _clickhouse_client()
-        try:
-            _drop_tables(client)
-            _assert_tables_absent(client)
-            client.command(
+        hook = _clickhouse_hook()
+        _drop_tables(hook)
+        _assert_tables_absent(hook)
+        hook.run(
+            [
                 f"""
                 CREATE TABLE default.{LOCAL_TABLE} ON CLUSTER {CLUSTER}
                 (
@@ -133,9 +119,7 @@ def test_clickhouse():
                     '{{replica}}'
                 )
                 ORDER BY marker
-                """
-            )
-            client.command(
+                """,
                 f"""
                 CREATE TABLE default.{DISTRIBUTED_TABLE} ON CLUSTER {CLUSTER}
                 AS default.{LOCAL_TABLE}
@@ -145,86 +129,77 @@ def test_clickhouse():
                     '{LOCAL_TABLE}',
                     cityHash64(marker)
                 )
-                """
-            )
-            for node_name, source in NODES:
-                actual_tables = _table_engines(client, source)
-                if actual_tables != EXPECTED_TABLES:
-                    raise RuntimeError(
-                        f"неверный набор таблиц на {node_name}: {actual_tables}"
-                    )
-        finally:
-            client.close()
+                """,
+            ],
+        )
+        for node_name, source in NODES:
+            actual_tables = _table_engines(hook, source)
+            if actual_tables != EXPECTED_TABLES:
+                raise RuntimeError(
+                    f"неверный набор таблиц на {node_name}: {actual_tables}"
+                )
 
     @task
     def write_marker() -> dict[str, str]:
-        client = _clickhouse_client()
-        try:
-            marker = f"{get_current_context()['run_id']}:{uuid.uuid4()}"
-            client.insert(
-                f"default.{LOCAL_TABLE}",
-                [[marker]],
-                column_names=["marker"],
-            )
-            local_rows = client.query(
-                f"""
-                SELECT hostName(), marker
-                FROM default.{LOCAL_TABLE}
-                WHERE marker = {{marker:String}}
-                """,
-                parameters={"marker": marker},
-            ).result_rows
-            if len(local_rows) != 1 or local_rows[0][1] != marker:
-                raise RuntimeError(f"маркер не найден в локальной таблице: {marker}")
-            return {"marker": marker, "hostname": local_rows[0][0]}
-        finally:
-            client.close()
+        hook = _clickhouse_hook()
+        marker = f"{get_current_context()['run_id']}:{uuid.uuid4()}"
+        hook.bulk_insert_rows(
+            f"default.{LOCAL_TABLE}",
+            [[marker]],
+            column_names=["marker"],
+        )
+        local_rows = hook.get_records(
+            f"""
+            SELECT hostName(), marker
+            FROM default.{LOCAL_TABLE}
+            WHERE marker = {{marker:String}}
+            """,
+            parameters={"marker": marker},
+        )
+        if len(local_rows) != 1 or local_rows[0][1] != marker:
+            raise RuntimeError(f"маркер не найден в локальной таблице: {marker}")
+        return {"marker": marker, "hostname": local_rows[0][0]}
 
     @task
     def read_from_node_2(written: dict[str, str]) -> None:
-        client = _clickhouse_client()
-        try:
-            node_2_rows = client.query(
-                """
-                SELECT hostName()
-                FROM remote(
-                    'clickhouse-02:9000', 'system', 'one',
-                    {remote_user:String}, {remote_password:String}
-                )
-                """,
-                parameters=_remote_parameters(),
-            ).result_rows
-            if len(node_2_rows) != 1:
-                raise RuntimeError(f"не удалось определить имя ноды 2: {node_2_rows}")
-            distributed_rows = client.query(
-                f"""
-                SELECT _shard_num, hostName(), marker
-                FROM remote(
-                    'clickhouse-02:9000',
-                    'default',
-                    '{DISTRIBUTED_TABLE}',
-                    {{remote_user:String}},
-                    {{remote_password:String}}
-                )
-                WHERE marker = {{marker:String}}
-                """,
-                parameters={
-                    "marker": written["marker"],
-                    **_remote_parameters(),
-                },
-            ).result_rows
-            if written["hostname"] == node_2_rows[0][0]:
-                raise RuntimeError(
-                    "запись и чтение маркера должны выполняться с разных нод"
-                )
-            expected_rows = [(1, written["hostname"], written["marker"])]
-            if distributed_rows != expected_rows:
-                raise RuntimeError(
-                    "нода 2 не прочитала маркер первого шарда через Distributed: "
-                    f"{written['marker']}, получено {distributed_rows}"
-                )
-        finally:
-            client.close()
+        hook = _clickhouse_hook()
+        node_2_rows = hook.get_records(
+            """
+            SELECT hostName()
+            FROM remote(
+                'clickhouse-02:9000', 'system', 'one',
+                {remote_user:String}, {remote_password:String}
+            )
+            """,
+            parameters=_remote_parameters(),
+        )
+        if len(node_2_rows) != 1:
+            raise RuntimeError(f"не удалось определить имя ноды 2: {node_2_rows}")
+        distributed_rows = hook.get_records(
+            f"""
+            SELECT _shard_num, hostName(), marker
+            FROM remote(
+                'clickhouse-02:9000',
+                'default',
+                '{DISTRIBUTED_TABLE}',
+                {{remote_user:String}},
+                {{remote_password:String}}
+            )
+            WHERE marker = {{marker:String}}
+            """,
+            parameters={
+                "marker": written["marker"],
+                **_remote_parameters(),
+            },
+        )
+        if written["hostname"] == node_2_rows[0][0]:
+            raise RuntimeError("запись и чтение маркера должны идти с разных нод")
+        expected_rows = [(1, written["hostname"], written["marker"])]
+        if distributed_rows != expected_rows:
+            raise RuntimeError(
+                "нода 2 не прочитала маркер первого шарда через Distributed: "
+                f"{written['marker']}, получено {distributed_rows}"
+            )
 
     # Уборка идёт только после успеха: упавший пробник оставляет кластер таким,
     # каким сломался, а остатки сносит начало следующего запуска. Правило
@@ -233,12 +208,9 @@ def test_clickhouse():
     # с упавшей проверкой (ADR 0003).
     @task
     def cleanup_tables() -> None:
-        client = _clickhouse_client()
-        try:
-            _drop_tables(client)
-            _assert_tables_absent(client)
-        finally:
-            client.close()
+        hook = _clickhouse_hook()
+        _drop_tables(hook)
+        _assert_tables_absent(hook)
 
     prepared = prepare_tables()
     written = write_marker()

dags/world_lifecycle.py

@@ -18,10 +18,12 @@ import time
 from pathlib import Path
 
 from airflow.exceptions import AirflowException
+from airflow.providers.clickhousedb.hooks.clickhouse import ClickHouseHook
+from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator
 from airflow.providers.docker.hooks.docker import DockerHook
 from airflow.providers.docker.operators.docker import DockerOperator
 from airflow.providers.standard.operators.trigger_dagrun import TriggerDagRunOperator
-from airflow.sdk import Connection, Variable, dag, task
+from airflow.sdk import Variable, chain, dag, task
 
 START_DATE = datetime.datetime(2026, 1, 1, tzinfo=datetime.UTC)
 TAGS = ["жизненный цикл мира"]
@@ -34,7 +36,8 @@ KAFKA_BOOTSTRAP_SERVERS = os.environ["KAFKA_BOOTSTRAP_SERVERS"]
 GENERATOR_IMAGE = os.environ["GENERATOR_IMAGE"]
 STAND_NETWORK = os.environ["STAND_NETWORK"]
 
-SQL_ROOT = Path("/opt/airflow/sql/ddl")
+SQL_ROOT = Path("/opt/airflow/sql")
+DDL_ROOT = SQL_ROOT / "ddl"
 WORLD_ROOT = Path("/opt/airflow/world")
 INVENTORY = WORLD_ROOT / "data/world-inventory.json"
 CLICKHOUSE_CONNECTION = "clickhouse_lifecycle"
@@ -44,48 +47,9 @@ CONSUMER_GROUPS = ("clickstream_hits", "clickstream_orders")
 TOPICS = {HITS_TOPIC: 2, ORDERS_TOPIC: 1}
 
 
-def _clickhouse_client():
-    """Открыть соединение владельца жизненного цикла."""
-    import clickhouse_connect
-
-    connection = Connection.get(CLICKHOUSE_CONNECTION)
-    return clickhouse_connect.get_client(
-        host=connection.host,
-        port=connection.port,
-        username=connection.login,
-        password=connection.password,
-        database=connection.schema or "default",
-        connect_timeout=5,
-        # `ON CLUSTER` ждёт хосты до 180 с. Транспорт живёт дольше, чтобы
-        # ClickHouse сам назвал незавершённый хост вместо сетевого тайм-аута.
-        send_receive_timeout=300,
-    )
-
-
-def _ddl_statements(text: str) -> list[str]:
-    """Разделить канонический DDL по его явному правилу файлов.
-
-    ClickHouse Connect исполняет один запрос за вызов. В `sql/ddl` конец
-    выражения — точка с запятой в конце строки; это проверяемая конвенция
-    наших файлов, а не попытка написать общий SQL-парсер.
-    """
-    statements: list[str] = []
-    lines: list[str] = []
-    for line in text.splitlines():
-        lines.append(line)
-        stripped = line.rstrip()
-        if stripped.endswith(";") and not stripped.lstrip().startswith("--"):
-            statement = "\n".join(lines).strip()
-            statements.append(statement[:-1].rstrip())
-            lines = []
-
-    remainder = "\n".join(lines).strip()
-    if remainder and any(
-        line.strip() and not line.lstrip().startswith("--")
-        for line in remainder.splitlines()
-    ):
-        raise AirflowException("DDL-файл оканчивается незавершённым выражением")
-    return statements
+def _clickhouse_hook() -> ClickHouseHook:
+    """Подключить владельца жизненного цикла."""
+    return ClickHouseHook(clickhouse_conn_id=CLICKHOUSE_CONNECTION)
 
 
 def _generator(task_id: str, command: list[str]) -> DockerOperator:
@@ -118,6 +82,7 @@ def _kafka_error_code(error: Exception) -> int:
     start_date=START_DATE,
     is_paused_upon_creation=False,
     max_active_runs=1,
+    template_searchpath=str(SQL_ROOT),
     tags=TAGS,
 )
 def world_initialize():
@@ -194,23 +159,6 @@ def world_initialize():
                     "выполните make rebuild-storage"
                 )
 
-    @task
-    def apply_ddl() -> None:
-        """Применить канонические файлы DDL с ноды 1 по порядку имён."""
-        files = sorted(SQL_ROOT.glob("*.sql"))
-        if not files:
-            raise AirflowException(f"в {SQL_ROOT} нет файлов DDL")
-
-        client = _clickhouse_client()
-        try:
-            for path in files:
-                statements = _ddl_statements(path.read_text(encoding="utf-8"))
-                logging.info("применяем %s: выражений %s", path.name, len(statements))
-                for statement in statements:
-                    client.command(statement)
-        finally:
-            client.close()
-
     @task
     def build_generator_image() -> None:
         """Собрать генератор из текущего дерева, а не из старого образа."""
@@ -248,22 +196,19 @@ def world_initialize():
               + (SELECT count() FROM ods.event_errors_dist)
         """
 
-        client = _clickhouse_client()
-        try:
-            arrived = 0
-            for attempt in range(1, 101):
-                row = client.query(query).first_row
-                if row is None:
-                    raise AirflowException("ClickHouse не вернул счётчик приёма")
-                arrived = int(row[0])
-                if arrived >= expected:
-                    logging.info("стартовый мир принят: строк %s", arrived)
-                    return
-                if attempt % 5 == 0:
-                    logging.info("принято %s из %s строк", arrived, expected)
-                time.sleep(3)
-        finally:
-            client.close()
+        hook = _clickhouse_hook()
+        arrived = 0
+        for attempt in range(1, 101):
+            row = hook.get_first(query)
+            if row is None:
+                raise AirflowException("ClickHouse не вернул счётчик приёма")
+            arrived = int(row[0])
+            if arrived >= expected:
+                logging.info("стартовый мир принят: строк %s", arrived)
+                return
+            if attempt % 5 == 0:
+                logging.info("принято %s из %s строк", arrived, expected)
+            time.sleep(3)
         raise AirflowException(
             f"стартовый мир не принят за 300 с: строк {arrived} из {expected}; "
             "проверьте журнал send_initial_events и "
@@ -279,7 +224,19 @@ def world_initialize():
     route = choose_path()
     ready = world_already_exists()
     topics = create_topics()
-    ddl = apply_ddl()
+    ddl_files = sorted(DDL_ROOT.glob("*.sql"))
+    if not ddl_files:
+        raise AirflowException(f"в {DDL_ROOT} нет файлов DDL")
+    ddl_tasks = [
+        SQLExecuteQueryOperator(
+            task_id=f"apply_{path.stem.replace('-', '_')}",
+            conn_id=CLICKHOUSE_CONNECTION,
+            sql=f"ddl/{path.name}",
+            split_statements=True,
+            do_xcom_push=False,
+        )
+        for path in ddl_files
+    ]
     image = build_generator_image()
     events = _generator(
         "send_initial_events",
@@ -307,7 +264,7 @@ def world_initialize():
     position = remember_starting_position()
 
     route >> [ready, topics]
-    topics >> ddl >> image >> events >> arrived >> snapshots >> orders >> position
+    chain(topics, *ddl_tasks, image, events, arrived, snapshots, orders, position)
 
 
 @dag(
@@ -328,16 +285,13 @@ def world_recreate():
 
     @task
     def drop_application_databases() -> None:
-        client = _clickhouse_client()
-        try:
-            for database in APPLICATION_DATABASES:
-                logging.info("удаляем базу %s", database)
-                client.command(
-                    f"DROP DATABASE IF EXISTS {database} "
-                    f"ON CLUSTER {CLICKHOUSE_CLUSTER} SYNC"
-                )
-        finally:
-            client.close()
+        hook = _clickhouse_hook()
+        for database in APPLICATION_DATABASES:
+            logging.info("удаляем базу %s", database)
+            hook.run(
+                f"DROP DATABASE IF EXISTS {database} "
+                f"ON CLUSTER {CLICKHOUSE_CLUSTER} SYNC"
+            )
 
     @task
     def delete_consumer_groups() -> None: