Переход с airflow-clickhouse-plugin на провайдер ClickHouse

Перенос существующих DAG Airflow с пакета сообщества airflow-clickhouse-plugin на официальный провайдер apache-airflow-providers-clickhousedb

До появления официального провайдера ClickHouse большинство пользователей Airflow подключались к ClickHouse через пакет сообщества airflow-clickhouse-plugin. В этом руководстве описан перенос существующего развертывания на apache-airflow-providers-clickhousedb. О том, как работает сам провайдер, см. Подключение Apache Airflow к ClickHouse.

​

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

Провайдеры — это механизм интеграции Airflow со сторонними системами. Они выпускаются, тестируются и документируются вместе со всей остальной экосистемой Airflow, а данный провайдер поддерживается сообществом Airflow совместно с командой ClickHouse. Он построен на clickhouse-connect — клиенте 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	Фиксированный набор ключей, см. Дополнительные параметры соединения
Операторы и сенсоры	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.

pip install apache-airflow-providers-clickhousedb

Эти два пакета находятся в разных пространствах имен Python, поэтому их можно установить одновременно и переносить DAG по одному. Удалите плагин, когда его больше ничто не импортирует:

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 предоставляет 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/<version> apache-airflow-providers-clickhousedb/<version> и добавляет ваше значение в виде 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.

До:

{
    "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
    }
}

После:

{
    "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:

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).

​

Шаг 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.
settings	hook_params={"session_settings": {...}}	Шаблонизируется через hook_params. Значения накладываются поверх session_settings соединения.
do_xcom_push	do_xcom_push	Тот же флаг, но для задач с несколькими операторами результат имеет иную форму. Плагин передавал только результат последнего оператора. Если передан список операторов, провайдер передаёт список с отдельной записью на каждый оператор. См. Результаты нескольких операторов.
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 вместе со строками. См. Сохранение типов столбцов.
external_tables	ClickHouseHook.get_client()	Используйте clickhouse_connect.driver.external.ExternalData вместе с client.query.
columnar	ClickHouseHook.get_client()	client.query(...).result_columns.
types_check	Удалить	Только для собственного протокола.

До:

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},
)

После:

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 оператор отправит только строки последнего оператора — как и плагин.

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:

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:

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-строку. Вместо этого используйте нативную вставку:

До:

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)

После:

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 аргументы, которые предоставлял плагин:

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

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

До:

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,
)

После:

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 содержит версии 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. Промпт, который хорошо себя показал:

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=<rows>) 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
• Клиент Python clickhouse-connect
• Справочная документация apache-airflow-providers-clickhousedb
• airflow-clickhouse-plugin на GitHub