> ## Documentation Index > Fetch the complete documentation index at: https://private-7c7dfe99-vortex-format.mintlify.site/llms.txt > Use this file to discover all available pages before exploring further. > Перенос существующих DAG Airflow с пакета сообщества airflow-clickhouse-plugin на официальный провайдер apache-airflow-providers-clickhousedb # Переход с airflow-clickhouse-plugin на провайдер ClickHouse До появления [официального провайдера ClickHouse](/ru/integrations/connectors/data-ingestion/etl-tools/airflow-and-clickhouse) большинство пользователей Airflow подключались к ClickHouse через пакет сообщества [airflow-clickhouse-plugin](https://github.com/bryzgaloff/airflow-clickhouse-plugin). В этом руководстве описан перенос существующего развертывания на `apache-airflow-providers-clickhousedb`. О том, как работает сам провайдер, см. [Подключение Apache Airflow к ClickHouse](/ru/integrations/connectors/data-ingestion/etl-tools/airflow-and-clickhouse).

Зачем нужен нативный провайдер?

Провайдеры — это механизм интеграции Airflow со сторонними системами. Они выпускаются, тестируются и документируются вместе со всей остальной экосистемой Airflow, а данный провайдер поддерживается сообществом Airflow совместно с командой ClickHouse. Он построен на [clickhouse-connect](/ru/integrations/language-clients/python/index) — клиенте Python, который разрабатывает и поддерживает сама компания ClickHouse, а не на драйвере, сопровождаемом сообществом. Благодаря этому новые возможности сервера и исправления доходят до пользователей Airflow по поддерживаемому пути. Переход на провайдер даёт вам пакет с официальным местом размещения, стандартные операторы и сенсоры `common.sql`, а также тип соединения, который отображается в интерфейсе Airflow так же, как и для любой другой базы данных. Эти два пакета различаются не только путями импорта. Плагин обращается к ClickHouse по **собственному TCP-протоколу** с помощью `clickhouse-driver`. Провайдер работает по **HTTP(S)** через `clickhouse-connect` и использует универсальные операторы `common.sql` вместо того, чтобы поставлять собственные, специфичные для ClickHouse. Прочитайте руководство целиком, прежде чем что-либо менять. В частности, изменение соединения затрагивает сразу все DAG.

Краткий обзор

| | `airflow-clickhouse-plugin` | `apache-airflow-providers-clickhousedb` | | - | - | - | | Корень импорта | `airflow_clickhouse_plugin` | `airflow.providers.clickhousedb` | | Драйвер | `clickhouse-driver` | `clickhouse-connect` | | Протокол / порт по умолчанию | Нативный TCP, `9000` (`9440` с TLS) | HTTP, `8123` (`8443` с TLS) | | Тип соединения | Не регистрируется; подходит любой тип | `clickhouse` | | Дополнительные параметры соединения | Передаются без изменений в `clickhouse_driver.Client` | Фиксированный набор ключей, см. [Дополнительные параметры соединения](/ru/integrations/connectors/data-ingestion/etl-tools/airflow-and-clickhouse#extra-connection-options) | | Операторы и сенсоры | `ClickHouseOperator`, `ClickHouseSensor`, а также обёртки `common.sql` с префиксом `ClickHouse` | Операторы и сенсоры `common.sql` используются напрямую | | Hook | `ClickHouseHook` (`BaseHook`) и `ClickHouseDbApiHook` (`DbApiHook`) | Один `ClickHouseHook` (`DbApiHook`) | | Минимальная версия Airflow | 2.0 | 2.11 |

Шаг 1: Проверьте prerequisites и выполните установку

Для работы provider требуется Airflow 2.11 или новее, а также `apache-airflow-providers-common-sql` 1.32.0 или новее. Если у вас более старый release, сначала обновите Airflow. ```bash theme={null} pip install apache-airflow-providers-clickhousedb ``` Эти два пакета находятся в разных пространствах имен Python, поэтому их можно установить одновременно и переносить DAG по одному. Удалите плагин, когда его больше ничто не импортирует: ```bash theme={null} pip uninstall airflow-clickhouse-plugin clickhouse-driver ```

Шаг 2. Обновите соединения

Именно на этом шаге всё ломается, если его пропустить. Все существующие соединения с ClickHouse указывают на нативный порт, а провайдеру нужен HTTP-порт. | Поле | Плагин | Провайдер | | - | - | - | | Тип соединения | Любой (часто `sqlite` или `generic`) | `clickhouse` | | Порт, без шифрования | `9000` | `8123` | | Порт, TLS | `9440` | `8443` | | Login | Значение драйвера по умолчанию `default` | То же | | Schema | База данных | То же | Если ClickHouse находится за firewall или load balancer, перед переключением убедитесь, что HTTP-порт доступен с воркеров. Проверьте, что HTTP interface включён на сервере (`http_port` или `https_port` в конфигурации сервера). [ClickHouse Cloud](/ru/products/cloud/getting-started/intro) предоставляет HTTPS только на порту `8443`. Соединения, сохранённые в виде URI (`clickhouse://user:pass@host:9000/db?secure=true`), уже имеют тип соединения `clickhouse`, поскольку Airflow определяет его по scheme URI. Для них меняются только порт и дополнительные ключи; значения из строки запроса разбираются как JSON, поэтому `secure=true` остаётся булевым значением.

Дополнительные параметры соединения

Плагин передавал каждый ключ из `extra` напрямую в `clickhouse_driver.Client`, поэтому соединения могут содержать любой именованный аргумент `clickhouse-driver`. Провайдер читает только фиксированный набор ключей, а всё остальное передаёт через `client_kwargs`. Соответствие следующее: | Extra плагина (`clickhouse-driver`) | Extra провайдера | Примечания | | - | - | - | | `secure` | `secure` | Без изменений. Не забудьте также изменить порт. | | `verify` | `verify` | Без изменений. | | `settings` | `session_settings` | То же содержимое, новый ключ. | | `compression` | `compress` | Булевый. Названия алгоритмов у драйверов различаются; `true` — безопасный выбор. | | `connect_timeout` | `connect_timeout` | Без изменений. | | `send_receive_timeout` | `send_receive_timeout` | Без изменений. | | `client_name` | `client_name` | Семантика изменилась. Провайдер всегда отправляет `apache-airflow/ apache-airflow-providers-clickhousedb/` и добавляет ваше значение в виде label. | | `ca_certs` | `client_kwargs.ca_cert` | Путь к набору CA. | | `certfile` / `keyfile` | `client_kwargs.client_cert` / `client_kwargs.client_cert_key` | Взаимный TLS. | | `server_hostname` | `client_kwargs.server_host_name` | Переопределение SNI для TLS. | | `alt_hosts`, `round_robin` | Нет эквивалента | Клиент подключается только к одному хосту, указанному в соединении. Если вы использовали список failover, направьте соединение на свой load balancer или конечную точку ClickHouse Cloud. | | `sync_request_timeout`, `tcp_keepalive`, `compress_block_size` | Удалить | Только для собственного протокола. | | `ssl_version`, `ciphers`, `use_numpy`, `client_revision`, `settings_is_important`, `opentelemetry_traceparent`, `opentelemetry_tracestate` | Удалить | Нет эквивалента в `clickhouse-connect`. | До: ```json theme={null} { "conn_type": "sqlite", "host": "ch.example.com", "port": 9440, "login": "airflow", "password": "secret", "schema": "analytics", "extra": { "secure": true, "settings": {"max_execution_time": 300}, "compression": true } } ``` После: ```json theme={null} { "conn_type": "clickhouse", "host": "ch.example.com", "port": 8443, "login": "airflow", "password": "secret", "schema": "analytics", "extra": { "secure": true, "session_settings": {"max_execution_time": 300}, "compress": true } } ``` Проверьте каждое перенесённое соединение, прежде чем вносить изменения в DAG: ```bash theme={null} airflow connections test clickhouse_default ```

Шаг 3: Замените импорты

| Класс плагина | Замена из provider | | - | - | | `airflow_clickhouse_plugin.hooks.clickhouse.ClickHouseHook` | `airflow.providers.clickhousedb.hooks.clickhouse.ClickHouseHook` | | `airflow_clickhouse_plugin.hooks.clickhouse_dbapi.ClickHouseDbApiHook` | `airflow.providers.clickhousedb.hooks.clickhouse.ClickHouseHook` | | `airflow_clickhouse_plugin.operators.clickhouse.ClickHouseOperator` | `airflow.providers.common.sql.operators.sql.SQLExecuteQueryOperator` | | `airflow_clickhouse_plugin.sensors.clickhouse.ClickHouseSensor` | `airflow.providers.common.sql.sensors.sql.SqlSensor` | | `airflow_clickhouse_plugin.operators.clickhouse_dbapi.ClickHouseSQLExecuteQueryOperator` | `airflow.providers.common.sql.operators.sql.SQLExecuteQueryOperator` | | `...clickhouse_dbapi.ClickHouseSQLCheckOperator` | `...common.sql.operators.sql.SQLCheckOperator` | | `...clickhouse_dbapi.ClickHouseSQLValueCheckOperator` | `...common.sql.operators.sql.SQLValueCheckOperator` | | `...clickhouse_dbapi.ClickHouseSQLIntervalCheckOperator` | `...common.sql.operators.sql.SQLIntervalCheckOperator` | | `...clickhouse_dbapi.ClickHouseSQLThresholdCheckOperator` | `...common.sql.operators.sql.SQLThresholdCheckOperator` | | `...clickhouse_dbapi.ClickHouseSQLColumnCheckOperator` | `...common.sql.operators.sql.SQLColumnCheckOperator` | | `...clickhouse_dbapi.ClickHouseSQLTableCheckOperator` | `...common.sql.operators.sql.SQLTableCheckOperator` | | `...clickhouse_dbapi.ClickHouseBranchSQLOperator` | `...common.sql.operators.sql.BranchSQLOperator` | | `airflow_clickhouse_plugin.sensors.clickhouse_dbapi.ClickHouseSqlSensor` | `airflow.providers.common.sql.sensors.sql.SqlSensor` | Обёртки `common.sql` с префиксом `ClickHouse` были нужны лишь для того, чтобы подставить hook плагина. Provider регистрирует тип соединения `clickhouse`, поэтому классы `common.sql` без префикса сами получают hook из соединения. Если вы использовали эти обёртки, миграция обычно сводится к правке строки импорта, удалению префикса `ClickHouse` и явной передаче `conn_id` (см. [шаг 7](#step-7-the-commonsql-wrapper-family)).

Шаг 4: от `ClickHouseOperator` к `SQLExecuteQueryOperator`

| Аргумент `ClickHouseOperator` | Эквивалент в `SQLExecuteQueryOperator` | Примечания | | - | - | - | | `sql` | `sql` | Строка, список строк или путь к файлу `.sql`. Шаблонизируется в обоих случаях. | | `clickhouse_conn_id` | `conn_id` | В плагине по умолчанию использовалось значение `clickhouse_default`. У `SQLExecuteQueryOperator` значения по умолчанию нет: передавайте `conn_id` в каждой задаче или задайте его один раз в `default_args`. | | `database` | `database` | Без изменений. | | `parameters` для `SELECT` | `parameters` | Плейсхолдеры `%(name)s` продолжают работать. Также доступна привязка на стороне сервера в виде `{name:Type}`. | | `parameters` для `INSERT` (список строк) | `SQLInsertRowsOperator` | Передайте строки (или XCom) как `rows`, укажите имена столбцов через `columns` и задайте `insert_args={"executemany": True}`, чтобы `clickhouse-connect` использовал нативную вставку. Как вариант, вызовите `ClickHouseHook.bulk_insert_rows` из `@task`, см. [шаг 5](#step-5-clickhousehookexecute-to-dbapihook-methods). | | `settings` | `hook_params={"session_settings": {...}}` | Шаблонизируется через `hook_params`. Значения накладываются поверх `session_settings` соединения. | | `do_xcom_push` | `do_xcom_push` | Тот же флаг, но для задач с несколькими операторами результат имеет иную форму. Плагин передавал только результат последнего оператора. Если передан список операторов, провайдер передаёт список с отдельной записью на каждый оператор. См. [Результаты нескольких операторов](#multi-statement-results). | | `query_id` | `hook_params={"session_settings": {"query_id": "..."}}` | Шаблонизируется через `hook_params`. Как и в плагине, один и тот же идентификатор отправляется для каждого оператора в задаче с несколькими операторами. Если значение не задано, `clickhouse-connect` генерирует уникальный идентификатор для каждого оператора; `handler` может получить последний из `cursor.summary[-1]["query_id"]`. | | `with_column_types` | `handler` | Передайте handler, который считывает `cursor.description` вместе со строками. См. [Сохранение типов столбцов](#keeping-column-types). | | `external_tables` | `ClickHouseHook.get_client()` | Используйте `clickhouse_connect.driver.external.ExternalData` вместе с `client.query`. | | `columnar` | `ClickHouseHook.get_client()` | `client.query(...).result_columns`. | | `types_check` | Удалить | Только для собственного протокола. | До: ```python theme={null} from airflow_clickhouse_plugin.operators.clickhouse import ClickHouseOperator update_income_aggregate = ClickHouseOperator( task_id="update_income_aggregate", clickhouse_conn_id="clickhouse_test", database="default", sql=( """ INSERT INTO aggregate SELECT eventDt, sum(price * qty) AS income FROM sales WHERE eventDt = '{{ ds }}' GROUP BY eventDt """, """ SELECT sum(income) FROM aggregate WHERE eventDt BETWEEN '{{ data_interval_start | ds }}' AND '{{ data_interval_end | ds }}' """, ), settings={"max_execution_time": 600}, ) ``` После: ```python theme={null} from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator update_income_aggregate = SQLExecuteQueryOperator( task_id="update_income_aggregate", conn_id="clickhouse_test", database="default", sql=[ """ INSERT INTO aggregate SELECT eventDt, sum(price * qty) AS income FROM sales WHERE eventDt = '{{ ds }}' GROUP BY eventDt """, """ SELECT sum(income) FROM aggregate WHERE eventDt BETWEEN '{{ data_interval_start | ds }}' AND '{{ data_interval_end | ds }}' """, ], hook_params={"session_settings": {"max_execution_time": 600}}, ) ```

Результаты нескольких операторов

Плагин отправлял в XCom результат **последнего** оператора. `SQLExecuteQueryOperator` возвращает по одному результату на каждый оператор, если `sql` задан списком, поэтому приведённый выше пример отправляет `[[], [(12345.0,)]]` вместо `[(12345.0,)]`, которые отправлял плагин. Выберите один из вариантов: * Изменить нижестоящий `xcom_pull` так, чтобы он брал последний элемент. * Передать операторы одной строкой, разделив их символом `;`, и задать `split_statements=True`. Тогда при значении по умолчанию `return_last=True` оператор отправит только строки последнего оператора — как и плагин. ```python theme={null} SQLExecuteQueryOperator( task_id="update_income_aggregate", conn_id="clickhouse_test", sql=""" INSERT INTO aggregate SELECT eventDt, sum(price * qty) AS income FROM sales WHERE eventDt = '{{ ds }}' GROUP BY eventDt; SELECT sum(income) FROM aggregate WHERE eventDt = '{{ ds }}' """, split_statements=True, ) ``` `ClickHouseOperator`, который вставлял список строк через `parameters`, становится `SQLInsertRowsOperator`: ```python theme={null} from airflow.providers.common.sql.operators.sql import SQLInsertRowsOperator load_rows = SQLInsertRowsOperator( task_id="load_rows", conn_id="clickhouse_default", table_name="some_ch_table", columns=["id", "name"], rows=extract_task.output, insert_args={"executemany": True}, ) ``` Всегда передавайте `columns`; без него оператор будет искать таблицу через SQLAlchemy.

Сохранение типов столбцов

`with_column_types=True` возвращал `(rows, [(name, type), ...])`. Это поведение можно воспроизвести с помощью handler; курсор `clickhouse-connect` возвращает имена типов ClickHouse в `cursor.description`: ```python theme={null} def fetch_with_column_types(cursor): return cursor.fetchall(), [(col[0], col[1]) for col in cursor.description] SQLExecuteQueryOperator( task_id="typed_query", conn_id="clickhouse_default", sql="SELECT id, name FROM users LIMIT 10", handler=fetch_with_column_types, ) ```

Шаг 5: от `ClickHouseHook.execute` к методам `DbApiHook`

Hook плагина предоставлял единственный метод — `execute`, повторяющий `clickhouse_driver.Client.execute`. Hook провайдера является `DbApiHook`, поэтому он получает стандартные методы, доступные у любого другого SQL-провайдера: `run`, `get_records`, `get_first`, `get_pandas_df`, `get_df`, `insert_rows` и `test_connection`. Аргументы конструктора `clickhouse_conn_id` и `database` остались без изменений. | Плагин | Провайдер | | - | - | | `hook.execute("SELECT ...")` | `hook.get_records("SELECT ...")` | | `hook.execute("SELECT ...", params={"d": ds})` | `hook.get_records("SELECT ...", parameters={"d": ds})` | | `hook.execute("SELECT count() ...")[0][0]` | `hook.get_first("SELECT count() ...")[0]` | | `hook.execute("INSERT INTO t VALUES", rows)` | `hook.bulk_insert_rows("t", rows, column_names=[...])` | | `hook.execute(["SET ...", "INSERT ...", "SELECT ..."])` | `hook.run([...], handler=fetch_all_handler, return_last=True)` | | `hook.execute(..., settings={...})` | `ClickHouseHook(session_settings={...})` | | `hook.execute(..., external_tables=..., columnar=..., query_id=...)` | `hook.get_client().query(...)` | | `hook.get_conn()`, возвращающий `clickhouse_driver.Client` | `hook.get_client()`, возвращающий `clickhouse_connect` `Client` | `fetch_all_handler` и остальные handlers импортируются из `airflow.providers.common.sql.hooks.handlers`. Самый распространённый приём работы с hook в коде эпохи плагина — массовая вставка. Обычным вызовом `run` её не выполнить: курсор DB-API попытается подставить строки в SQL-строку. Вместо этого используйте нативную вставку: До: ```python theme={null} from airflow_clickhouse_plugin.hooks.clickhouse import ClickHouseHook def sqlite_to_clickhouse(): records = SqliteHook().get_records("SELECT id, name FROM some_sqlite_table") ClickHouseHook().execute("INSERT INTO some_ch_table VALUES", records) ``` После: ```python theme={null} from airflow.providers.clickhousedb.hooks.clickhouse import ClickHouseHook def sqlite_to_clickhouse(): records = SqliteHook().get_records("SELECT id, name FROM some_sqlite_table") ClickHouseHook().bulk_insert_rows( "some_ch_table", records, column_names=["id", "name"], batch_size=100_000 ) ``` `bulk_insert_rows` требует `column_names`. `batch_size` необязателен и ограничивает потребление памяти при очень больших объёмах входных данных. Универсальный метод `insert_rows(table, rows, target_fields=[...], executemany=True)` также завершается нативной вставкой, но только при `executemany=True`; по умолчанию на каждую строку отправляется отдельный HTTP-запрос. Для всего, что не покрывается интерфейсом DB-API, `get_client()` возвращает необработанный клиент `clickhouse-connect`, настроенный на основе соединения Airflow. Он заменяет собой все специфичные для `clickhouse-driver` аргументы, которые предоставлял плагин: ```python theme={null} from clickhouse_connect.driver.external import ExternalData hook = ClickHouseHook() with hook.get_client() as client: ext = ExternalData(file_name="ids", structure="id UInt64", data=b"1\n2\n3\n") result = client.query( "SELECT * FROM events WHERE id IN ids", external_data=ext, settings={"query_id": "my-traceable-id"}, ) columns = result.result_columns ```

Шаг 6: с `ClickHouseSensor` на `SqlSensor`

Это единственная замена, при которой меняются входные данные вызываемого объекта. | | `ClickHouseSensor` | `SqlSensor` | | - | - | - | | Вызываемый объект успеха | `is_success(result)` | `success(cell)` | | Вызываемый объект неудачи | `is_failure(result)` | `failure(cell)` | | Что получает вызываемый объект | Весь результат последнего оператора — `list` из row tuples | По умолчанию первый столбец первой строки (`selector=itemgetter(0)`) | | Успех по умолчанию | `bool(result)`, true, если вернулась хотя бы одна строка | `bool(первая cell)`, если строки вернулись, и `False`, если нет | | `sql` | Строка или список команд, а также все аргументы `ClickHouseOperator` | Одна строка. Используйте `hook_params` для базы данных и настроек сеанса. | | Идентификатор соединения | `clickhouse_conn_id`, по умолчанию `clickhouse_default` | `conn_id`, обязателен | | Строки не возвращены | `is_success([])`, `False` при значении по умолчанию | `False` либо ошибка при `fail_on_empty=True` | Поскольку плагин передавал весь результирующий набор, рабочий код сенсора обращается к нему по индексу. При миграции уберите такое индексирование: До: ```python theme={null} from airflow_clickhouse_plugin.sensors.clickhouse import ClickHouseSensor ClickHouseSensor( task_id="poke_events_count", database="monitor", sql="SELECT count() FROM warnings WHERE eventDate = '{{ ds }}'", is_success=lambda result: result[0][0] > 10000, ) ``` После: ```python theme={null} from airflow.providers.common.sql.sensors.sql import SqlSensor SqlSensor( task_id="poke_events_count", conn_id="clickhouse_default", hook_params={"database": "monitor"}, sql="SELECT count() FROM warnings WHERE eventDate = '{{ ds }}'", success=lambda cnt: cnt > 10000, ) ``` Если вызываемому объекту нужна вся строка, передайте `selector=lambda row: row`. Если нужны все строки, реализуйте проверку на SQL так, чтобы запрос возвращал одно булевое значение или количество.

Шаг 7: семейство обёрток `common.sql`

Код, использовавший `ClickHouseSQLExecuteQueryOperator`, `ClickHouseSqlSensor` и другие обёртки с префиксом `ClickHouse`, потребует минимальных изменений: * Измените импорт на модуль `common.sql` и уберите префикс `ClickHouse` из имени класса. * Передавайте `conn_id` явно. Обёртки воспринимали отсутствующий или равный `None` `conn_id` как `clickhouse_default`; у классов `common.sql` значения по умолчанию нет, и без него они завершаются с ошибкой. Параметр `default_args={"conn_id": "clickhouse_default"}` закрывает весь DAG. * `database=` у операторов и `hook_params={"schema": ...}` у сенсора продолжают работать: hook провайдера воспринимает `schema` как алиас `database`. * `ClickHouseDbApiHook` становится `ClickHouseHook`. Аргумент конструктора `schema` по-прежнему принимается как алиас; `database` — нативное написание для ClickHouse, и при указании обоих приоритет остаётся за ним. * Соединению по-прежнему необходимы изменения порта и extras из шага 2 — обёртки тоже использовали собственный протокол.

Различия в поведении, требующие проверки

Даже после успешной компиляции кода некоторые вещи во время выполнения ведут себя иначе. **Значение XCom для задач INSERT.** Плагин передавал то, что возвращал `clickhouse-driver`, а для вставки `VALUES` с параметрами это было количество вставленных строк. Провайдер передаёт пустой результирующий набор для операторов, которые не возвращают строк. Нижестоящие задачи, читающие количество строк из XCom, должны получать его иначе — например, с помощью последующего `SELECT count()`. **Настройки сеанса и операторы `SET`.** Оба пакета выполняют список из нескольких операторов по одному соединению, а `clickhouse-connect` по умолчанию создаёт отдельный сеанс для каждого клиента, поэтому оператор `SET` в начале списка всё равно должен применяться к последующим операторам. И всё же предпочтительнее использовать `session_settings`: этот способ явный, поддерживает шаблонизацию и работает одинаково независимо от того, сохраняется ли сеанс на сервере или в промежуточном proxy. Если ваши DAG зависят от `SET`, проверьте поведение в своей среде. **Исключения.** Теперь ошибки представлены классами `clickhouse_connect.driver.exceptions.DatabaseError`, `OperationalError` или `ProgrammingError` вместо `clickhouse_driver.errors.ServerException` и `NetworkError`. Обновите блоки `except`, код `on_failure_callback` и логику повторных попыток, которая анализирует типы исключений. **Сжатие.** `clickhouse-driver` не включал сжатие, если не был задан параметр `compression`; `clickhouse-connect` по умолчанию включает сжатие HTTP-ответов и согласует алгоритм с сервером. Чтобы вернуть прежнее поведение, укажите `"compress": false` в дополнительных параметрах соединения. Пакет `clickhouse-cityhash`, который требовался плагину для сжатия по собственному протоколу, больше не нужен; `lz4` остаётся установленным как зависимость `clickhouse-connect`. **Сопоставление типов.** Оба драйвера возвращают нативные типы Python, но это разные кодовые базы. Проверьте задачи, зависящие от точных типов для `DateTime64` с часовыми поясами, `Decimal`, `UUID`, столбцов `Nullable` и вложенных значений `Array` или `Map`, особенно там, где результат передаётся в XCom и используется далее по цепочке. **Идентификация запросов в `system.query_log`.** Запросы теперь поступают через HTTP interface, поэтому отображаются со значением `interface = 2` вместо `1`, а столбец `http_user_agent` таблицы [`system.query_log`](/ru/reference/system-tables/query_log) содержит версии Airflow и провайдера, а также дополнительный параметр `client_name`, если он задан. Любой мониторинг, фильтровавший по собственному протоколу или по имени клиента `clickhouse-driver`, потребует обновления. Запрос `SELECT`, не возвращающий строк, создаёт вторую запись: курсор DB-API выполняет `SELECT * FROM (...) LIMIT 0`, чтобы получить метаданные столбцов. **Тайм-ауты по HTTP.** `send_receive_timeout` теперь соответствует тайм-ауту чтения HTTP, а любой proxy или load balancer между воркерами и ClickHouse применяет к запросу собственный idle timeout. Для операторов, которые выполнялись много минут по собственному протоколу, эти лимиты может потребоваться увеличить. **Работа с соединениями.** Hook создаёт клиент `clickhouse-connect` на каждый вызов `run` или `get_client` — так же, как плагин открывал новое native connection на каждый `execute`. Клиенты используют общий HTTP connection pool в рамках процесса, поэтому вызов `close()` для клиента, полученного через `get_client()`, — скорее хорошая практика, чем необходимость; пул освобождается при завершении процесса задачи. Client является менеджером контекста, поэтому `with hook.get_client() as client:` — самая аккуратная форма.

Контрольный список

1. Airflow версии 2.11 или новее. 2. HTTP-порт доступен с воркеров; TLS-сертификаты действительны для HTTP-конечной точки. 3. Для каждого соединения с ClickHouse: тип `clickhouse`, порт `8123` или `8443`, extras преобразованы согласно шагу 2, проверка выполнена командой `airflow connections test`. 4. Импорты заменены согласно шагу 3; префиксы `ClickHouse` убраны из обёрток `common.sql`. 5. `clickhouse_conn_id` переименован в `conn_id` у операторов и сенсоров, а `conn_id` задан для каждой задачи, которая полагалась на значение по умолчанию из плагина. 6. `settings=` перенесены в `hook_params={"session_settings": ...}`. 7. `hook.execute("INSERT ... VALUES", rows)` заменён на `bulk_insert_rows`; `ClickHouseOperator(parameters=rows)` заменён на `SQLInsertRowsOperator`. 8. Вызываемые объекты сенсоров переведены с `result[0][0]` на непосредственное значение ячейки. 9. Случаи использования `with_column_types`, `external_tables`, `columnar`, `query_id` и `types_check` переписаны с применением `handler` или `get_client()`. 10. Проверены нижестоящие потребители XCom из задач с несколькими операторами и задач INSERT. 11. Код, перехватывающий исключения `clickhouse_driver`, обновлён. 12. `airflow-clickhouse-plugin` и `clickhouse-driver` удалены.

Использование ИИ-ассистента для написания кода

Приведённое выше сопоставление намеренно сделано механическим, чтобы ИИ-ассистент мог применить его к репозиторию с DAG. Промпт, который хорошо себя показал: ```text theme={null} Migrate this repository from airflow-clickhouse-plugin to apache-airflow-providers-clickhousedb following https://clickhouse.com/docs/integrations/airflow/migrating-from-airflow-clickhouse-plugin - Replace every airflow_clickhouse_plugin import per the class mapping table. - Rename clickhouse_conn_id to conn_id on operators and sensors, not on hooks; add conn_id="clickhouse_default" wherever a task relied on the plugin's default. - Move settings= into hook_params={"session_settings": ...}. - Replace hook.execute(...) with get_records, get_first, run or bulk_insert_rows according to the hook table; never pass a list of rows as parameters. Replace ClickHouseOperator(parameters=) with SQLInsertRowsOperator. - Rewrite sensor callables to receive the first cell instead of the full result. - Flag, but do not silently rewrite, any use of with_column_types, external_tables, columnar, query_id or types_check, and any XCom consumer of an INSERT task. - List every Airflow connection that must change port and extras; do not edit connections yourself. Show the diff and a summary of items flagged for human review. ``` Просмотрите diff. Изменения в соединениях, семантика сенсоров и потребители XCom — именно те места, где автоматический рефакторинг чаще всего даёт сбой.

Дополнительные материалы

* [Подключение Apache Airflow к ClickHouse](/ru/integrations/connectors/data-ingestion/etl-tools/airflow-and-clickhouse) * [Клиент Python `clickhouse-connect`](/ru/integrations/language-clients/python/index) * [Справочная документация `apache-airflow-providers-clickhousedb`](https://airflow.apache.org/docs/apache-airflow-providers-clickhousedb/) * [airflow-clickhouse-plugin на GitHub](https://github.com/bryzgaloff/airflow-clickhouse-plugin)