Провайдер ClickHouse для Airflow — загрузка без дублей

 

В сентябре 2026 года ClickHouse выпустила ClickHouse provider for Apache Airflow (провайдер ClickHouse для Airflow), официальный пакет apache-airflow-providers-clickhousedb, и интеграция двух популярных инструментов дата-инженера впервые получила поддержку от самих разработчиков СУБД и сообщества Airflow. Новость короткая, а вопросы за ней длинные. Чем провайдер отличается от того, чем DAG ходили в ClickHouse раньше? Что происходит, когда задача Airflow падает после вставки и уходит в ретрай? Мы подняли стенд с Airflow 3.3.2 и ClickHouse 26.9, прогнали семь сценариев повторной загрузки и замерили три пути вставки, пакетные на миллионе строк. В статье разбираем устройство провайдера, механизм дедупликации вставок (insert deduplication) в MergeTree и одну неприятную находку, из-за которой провайдер теряет данные без всякого ретрая.

 

Что изменилось с выходом провайдера clickhousedb

До осени 2026 года официального пути из Airflow в ClickHouse не было. Команды писали PythonOperator с клиентом clickhouse-connect внутри или ставили сторонний пакет airflow-clickhouse-plugin, который работает поверх драйвера clickhouse-driver по нативному протоколу. Оба варианта живы, но первый каждая команда собирает заново, а второй сопровождается одним автором вне экосистемы Apache.

Анонс в блоге ClickHouse от 17 сентября закрывает этот пробел. Пакет apache-airflow-providers-clickhousedb версии 1.0.1 вышел на PyPI 13 октября под лицензией Apache-2.0. Он требует Python 3.11 или новее и Airflow начиная с 2.11.0, так что подходит и для ветки 2.x, и для Airflow 3. Короткий разбор самого релиза мы уже публиковали в новости о провайдере ClickHouse для Airflow. Здесь идём глубже и смотрим, что провайдер делает под капотом и где его поведение расходится с ожиданиями.

Если базовая связка Airflow и ClickHouse вам пока незнакома, начните с урока про пакетную загрузку в ClickHouse из Airflow. Там разобраны DAG, хук и типовой ETL-конвейер, а эта статья опирается на них и добавляет вопрос идемпотентности.

 

Как устроен провайдер

Провайдер маленький, и это его сильная сторона. В пакете один модуль хука, тип соединения и пример DAG. Мы разобрали исходник wheel 1.0.1, потому что документация описывает соединение, но молчит о том, как хук ведёт себя при вставке.

 

Соединение и session_settings

Класс ClickHouseHook наследует DbApiHook из провайдера common-sql и создаёт клиент clickhouse-connect по HTTP-интерфейсу ClickHouse. Порт по умолчанию 8123, для TLS используется 8443 с флагом secure в extra соединения. Тип соединения называется clickhouse, идентификатор по умолчанию clickhouse_default, поле schema задаёт базу данных.

Настройки ClickHouse передаются через словарь session_settings. Его можно положить в extra соединения или передать в конструктор хука, и значения из конструктора перекрывают значения из extra. Хук передаёт итоговый словарь в параметр settings клиента clickhouse-connect, поэтому он едет с каждым запросом этого клиента. Отдельно хук подписывает запросы в заголовке User-Agent версиями Airflow и провайдера, и по этой подписи в system.query_log легко найти всё, что пришло из DAG.

 

Почему нет своего оператора

Своего оператора в провайдере нет. Хук реализует DB-API 2.0 через тонкий адаптер, и этого хватает штатному SQLExecuteQueryOperator из common-sql с шаблонами Jinja, обработчиками результата и разбиением на отдельные выражения. Для DDL, проверок и выборок такой подход удобен, ведь новый синтаксис учить не нужно. Для загрузки данных путь DB-API хуже, и в замере ниже видно, насколько медленнее построчный insert_rows из common-sql.

 

bulk_insert_rows и insert context

Пакетная вставка живёт в методе bulk_insert_rows(table, rows, column_names, batch_size). Без batch_size хук отправляет все строки одним вызовом client.insert в колоночном формате clickhouse-connect. С batch_size он один раз создаёт insert context, чтобы не повторять DESCRIBE TABLE, и шлёт данные пачками заданного размера.

Здесь кроется деталь, важная для всего дальнейшего. Параметра settings у bulk_insert_rows нет. Любая настройка вставки, включая токен дедупликации, может прийти только из session_settings хука, а значит она одна на весь вызов и на все пачки внутри него. Если нужна настройка на конкретную вставку, приходится брать сырой клиент методом get_client() и закрывать его самостоятельно.

 

Apache Airflow для инженеров данных

Код курса

AIRF

Ближайшая дата курса

9 ноября, 2026

Продолжительность

24 ак.часов

Стоимость обучения

76 800

 

Почему ретраи Airflow дублируют данные в ClickHouse

Провайдер решает вопрос «чем подключиться». Вопрос «как не загрузить одно и то же дважды» он оставляет пользователю, и в примерах провайдера про него нет ни слова.

 

Ретрай задачи как доставка at-least-once

Задача Airflow с параметром retries перезапускается целиком, если первая попытка завершилась ошибкой. Airflow не знает, на какой строке кода случился сбой. Вставка в ClickHouse могла пройти успешно, а упасть мог следующий шаг, например запись метки в другую систему или таймаут воркера. Повторная попытка выполнит вставку ещё раз. Это классическая доставка at-least-once, при которой данные гарантированно доходят, но могут прийти повторно.

На скриншоте с нашего стенда именно этот сценарий. В логе первой попытки хук пишет «Inserted 1000 rows», сразу после этого задача падает с RuntimeError, а через пять секунд Airflow запускает вторую попытку, которая завершается успешно. По той же схеме падают и перезапускаются все семь задач загрузки в нашем DAG. На скриншоте задача с пачками, и сколько её строк реально осталось в таблице, разберём в практической части.

 

Дедупликация вставок в MergeTree

ClickHouse умеет отбрасывать повторные вставки сам. Сервер считает хеш каждого вставленного блока данных и хранит последние хеши в логе дедупликации (deduplication log). Если новый блок совпал с уже записанным, он не пишется. Механизм описан в руководстве ClickHouse по дедупликации вставок при ретраях, и у него два условия, о которых легко забыть.

• Окно дедупликации (deduplication window) должно быть ненулевым. У движков семейства ReplicatedMergeTree лог включён по умолчанию, на нашем стенде replicated_deduplication_window равен 10000. У обычного MergeTree параметр non_replicated_deduplication_window по умолчанию равен 0, и такая таблица не дедуплицирует ничего.
• Данные должны совпадать побайтно. Хеш считается по содержимому блока. Если в строках есть время загрузки, номер попытки или случайный идентификатор, повтор даёт другой хеш и проходит как новые данные.

Второе условие в реальных DAG нарушается постоянно, потому что колонка loaded_at со значением now() есть почти в каждой таблице сырого слоя.

 

insert_deduplication_token и async_insert

Для таких случаев есть настройка insert_deduplication_token. Если она задана, сервер берёт для дедупликации токен вместо хеша данных. Повтор с тем же токеном отбрасывается, даже если содержимое строк изменилось. Для Airflow естественный токен собирается из идентификаторов DAG, запуска и задачи. Он одинаков у первой попытки и у ретрая, но различается между запусками. Номер попытки и текущее время в токен класть нельзя.

Второй фактор появился в ClickHouse 26.2, где асинхронные вставки (asynchronous inserts) включены по умолчанию. При async_insert равном 1 сервер складывает мелкие вставки в буфер и сбрасывает их пачкой, а wait_for_async_insert равный 1 заставляет клиента ждать этого сброса. По документации дедупликация синхронных и асинхронных вставок с 26.2 использует общий лог, а старая настройка async_insert_deduplicate учитывается только в режиме обратной совместимости. Мы проверили это на стенде, результат ниже.

Построение DWH на ClickHouse

Код курса

CLICH

Ближайшая дата курса

30 ноября, 2026

Продолжительность

24 ак.часов

Стоимость обучения

76 800

 

Официальный провайдер и альтернативы

Теперь, когда понятен механизм, сравним три способа грузить данные в ClickHouse из Airflow. Таблица собрана по PyPI, исходникам пакетов версий из нашего стенда и замеру на миллионе строк, который показан в практической части.

Критерий	apache-airflow-providers-clickhousedb 1.0.1	airflow-clickhouse-plugin 1.8.0	PythonOperator и clickhouse-connect
Протокол и порт	HTTP, 8123 и 8443 для TLS	нативный TCP через clickhouse-driver 0.2.11, 9000	HTTP, 8123 и 8443 для TLS
Сопровождение и лицензия	сообщество Airflow и команда ClickHouse, Apache-2.0	сторонний проект, MIT	свой код команды, clickhouse-connect под Apache-2.0
Версии Airflow и Python	Airflow от 2.11.0, Python от 3.11	Python от 3.10, на Airflow 3.3.2 работает с DeprecatedImportWarning	задаёт сама команда, требования clickhouse-connect мы не сверяли
Свои операторы	нет, используется SQLExecuteQueryOperator	ClickHouseOperator, ClickHouseSensor и SQL-операторы проверок	нет
Настройки на вставку	только session_settings хука или get_client()	параметр settings в execute и в операторе	параметр settings в client.insert
Вставка 1 млн строк, медиана из трёх	1.31 с, около 760 тыс. строк/с	3.97 с, около 252 тыс. строк/с	не мерили, по исходнику тот же client.insert, что у bulk_insert_rows

Официальный провайдер выигрывает в сопровождении и в скорости пакетной вставки. На миллионе строк bulk_insert_rows оказался втрое быстрее плагина. Профилированием причину мы не проверяли, но провайдер отдаёт данные в колоночном формате clickhouse-connect, а плагин идёт через clickhouse-driver. На 10 тысячах строк в пилотном замере картина была обратной, хотя туда попал первый вызов хука с DESCRIBE TABLE, поэтому выводы о скорости имеет смысл делать только на объёме, близком к боевому.

Плагин выигрывает в удобстве. У него есть свои операторы и сенсор, а settings передаются прямо в вызов, так что токен дедупликации на каждую вставку задаётся одной строкой. Ручной PythonOperator с clickhouse-connect по исходнику вызывает тот же client.insert, что и провайдер, поэтому скорость у него должна быть той же, хотя отдельно мы его не мерили. Контроль над settings у него тот же над settings, что у плагина, но соединения, ретраи подключения и подпись запросов придётся писать самим.

 

Практика на стенде, загрузка без дублей

По традиции весь код используемый в статье выкладываем на наш GitHub репозиторий

 

Стенд

Стенд работает на сервере AWS t3.xlarge с 4 vCPU и 16 ГБ памяти под Ubuntu 24.04. В Docker Compose два сервиса, ClickHouse 26.9.23.15 и Airflow 3.3.2 в режиме standalone, каждому выделено по 4 ГБ. Для провайдера заведено соединение clickhouse_default по HTTP, для плагина соединение clickhouse_native по порту 9000. Оба заданы переменными окружения AIRFLOW_CONN_*, так что настраивать что-либо в интерфейсе не нужно.

# ClickHouse 26.9.23.15 + Airflow 3.3.2 standalone, прогнано на стенде 2026-10-21
name: blog-airflow_clickhouse_provider

services:
  clickhouse:
    image: clickhouse/clickhouse-server:26.9.23.15
    mem_limit: 4g
    environment:
      CLICKHOUSE_DB: demo
      CLICKHOUSE_USER: airflow
      CLICKHOUSE_PASSWORD: airflow
      CLICKHOUSE_DEFAULT_ACCESS_MANAGEMENT: "1"
    ulimits:
      nofile: { soft: 262144, hard: 262144 }
    ports:
      - "127.0.0.1:8123:8123"   # HTTP, им ходит провайдер clickhousedb
      - "127.0.0.1:9000:9000"   # native, им ходит airflow-clickhouse-plugin
    volumes:
      - ch_data:/var/lib/clickhouse
    healthcheck:
      test: ["CMD", "clickhouse-client", "--user", "airflow", "--password", "airflow", "-q", "SELECT 1"]
      interval: 5s
      timeout: 5s
      retries: 30

  airflow:
    build: .
    image: blog-airflow_clickhouse_provider-airflow:3.3.2
    mem_limit: 4g
    command: standalone
    depends_on:
      clickhouse: { condition: service_healthy }
    environment:
      AIRFLOW__CORE__LOAD_EXAMPLES: "False"
      AIRFLOW__CORE__SIMPLE_AUTH_MANAGER_ALL_ADMINS: "True"
      AIRFLOW__CORE__DAGS_FOLDER: /opt/airflow/dags
      # соединение провайдера: HTTP 8123, тип clickhouse
      AIRFLOW_CONN_CLICKHOUSE_DEFAULT: '{"conn_type": "clickhouse", "host": "clickhouse", "port": 8123, "login": "airflow", "password": "airflow", "schema": "demo"}'
      # соединение плагина: native 9000, extra уходит в clickhouse_driver.Client как есть
      AIRFLOW_CONN_CLICKHOUSE_NATIVE: '{"conn_type": "generic", "host": "clickhouse", "port": 9000, "login": "airflow", "password": "airflow", "schema": "demo"}'
    ports:
      - "127.0.0.1:8080:8080"
    volumes:
      - ./dags:/opt/airflow/dags
      - ./:/opt/airflow/demo
    healthcheck:
      test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://localhost:8080/api/v2/monitor/health')"]
      interval: 10s
      timeout: 5s
      retries: 30
      start_period: 30s

volumes:
  ch_data:

Образ Airflow собирается из официального с доустановкой пакетов. Сам Airflow закреплён в той же команде pip, иначе установка провайдеров может потянуть его обновление.

# apache/airflow 3.3.2 + провайдер clickhousedb 1.0.1 и airflow-clickhouse-plugin 1.8.0, прогнано на стенде 2026-10-21
FROM apache/airflow:3.3.2-python3.12
COPY requirements.txt /requirements.txt
# пин самого Airflow, чтобы pip не обновил его вместе с провайдерами
RUN pip install --no-cache-dir "apache-airflow==3.3.2" -r /requirements.txt

apache-airflow-providers-clickhousedb==1.0.1
apache-airflow-providers-common-sql==2.3.0
clickhouse-connect==1.11.0
airflow-clickhouse-plugin==1.8.0
clickhouse-driver==0.2.11

Стенд поднимается одной командой, а версии пакетов внутри контейнера проверяем через pip freeze.

# Docker Compose v2, прогнано на стенде 2026-10-21
docker compose up -d --build --wait
docker compose exec -T -e AIRFLOW__LOGGING__LOGGING_LEVEL=WARNING airflow sh -c "python --version; pip freeze | grep -iE '^(apache-airflow|apache-airflow-providers-(clickhousedb|common-sql)|clickhouse-connect|clickhouse-driver|airflow-clickhouse-plugin)=='"

Python 3.12.14
airflow-clickhouse-plugin==1.8.0
apache-airflow==3.3.2
apache-airflow-providers-clickhousedb==1.0.1
apache-airflow-providers-common-sql==2.3.0
clickhouse-connect==1.11.0
clickhouse-driver==0.2.11

 

Соединение и настройки сервера

Первый скрипт проверяет соединение, читает версию сервера и дефолты, от которых зависит дедупликация. Заодно он убеждается, что session_settings действительно доходят до сервера, и достаёт из query_log подпись провайдера.

# apache-airflow-providers-clickhousedb 1.0.1, clickhouse-connect 1.11.0, ClickHouse 26.9.23.15, прогнано на стенде 2026-10-21
# Запуск внутри контейнера Airflow: соединение clickhouse_default берётся из AIRFLOW_CONN_CLICKHOUSE_DEFAULT
from airflow.providers.clickhousedb.hooks.clickhouse import ClickHouseHook

hook = ClickHouseHook()  # clickhouse_conn_id по умолчанию clickhouse_default

# test_connection() из DbApiHook выполняет SELECT 1 через DB-API адаптер провайдера
print("test_connection:", hook.test_connection())

# get_records идёт через курсор clickhouse_connect.dbapi, как в SQLExecuteQueryOperator
print("версия сервера:", hook.get_records("SELECT version()")[0][0])

# дефолты сервера, от которых зависит дедупликация вставок
print("настройки сессии:")
for name, value, changed in hook.get_records(
    "SELECT name, value, changed FROM system.settings WHERE name IN "
    "('async_insert', 'wait_for_async_insert', 'async_insert_deduplicate', "
    "'insert_deduplicate', 'insert_deduplication_token') ORDER BY name"
):
    print(f"  {name} = {value!r} (changed={changed})")

print("настройки MergeTree:")
for name, value in hook.get_records(
    "SELECT name, value FROM system.merge_tree_settings "
    "WHERE name IN ('non_replicated_deduplication_window', 'replicated_deduplication_window') ORDER BY name"
):
    print(f"  {name} = {value}")

# session_settings хука уходят с каждым запросом клиента
tuned = ClickHouseHook(session_settings={"max_threads": 2})
print("max_threads с session_settings:", tuned.get_records("SELECT getSetting('max_threads')")[0][0])

# провайдер подписывает запросы версиями Airflow и своей в User-Agent, это видно в query_log
client = hook.get_client()
client.command("SYSTEM FLUSH LOGS")
ua = client.query(
    "SELECT http_user_agent FROM system.query_log "
    "WHERE http_user_agent LIKE 'apache-airflow%' ORDER BY event_time_microseconds DESC LIMIT 1"
).result_rows
client.close()
print("http_user_agent:", ua[0][0] if ua else "нет записи")

# Docker Compose v2, прогнано на стенде 2026-10-21
docker compose exec -T -e AIRFLOW__LOGGING__LOGGING_LEVEL=WARNING airflow sh -c "cd /opt/airflow/demo && python check_connection.py"

test_connection: (True, 'Connection successfully tested')
версия сервера: 26.9.23.15
настройки сессии:
  async_insert = '1' (changed=0)
  async_insert_deduplicate = '0' (changed=0)
  insert_deduplicate = '1' (changed=0)
  insert_deduplication_token = '' (changed=0)
  wait_for_async_insert = '1' (changed=0)
настройки MergeTree:
  non_replicated_deduplication_window = 0
  replicated_deduplication_window = 10000
max_threads с session_settings: 2
http_user_agent: apache-airflow/3.3.2 apache-airflow-providers-clickhousedb/1.0.1 clickhouse-connect/1.11.0 (lv:py/3.12.14; mode:sync; os:linux; os_user:airflow)

Две строки вывода определяют всё дальнейшее. Первая это async_insert = ‘1’ при wait_for_async_insert = ‘1’, то есть сервер 26.9 по умолчанию буферизует вставки и заставляет клиента ждать сброса. Вторая это non_replicated_deduplication_window = 0, обычный MergeTree без явной настройки ничего не дедуплицирует. Значение max_threads равное 2 подтверждает, что настройки из конструктора хука доходят до сервера.

 

Ретрай без токена и с токеном

DAG retry_dedup создаёт семь таблиц и грузит в каждую по 1000 строк. Колонка loaded_at заполняется текущим временем, поэтому хеш блока у повтора всегда другой. Каждая задача загрузки после успешной вставки падает на первой попытке и через пять секунд уходит в ретрай. Таблицы различаются тремя параметрами, а именно окном дедупликации, токеном и режимом вставки. Задача report в конце считает строки.

# apache-airflow 3.3.2, apache-airflow-providers-clickhousedb 1.0.1, ClickHouse 26.9.23.15, прогнано на стенде 2026-10-21
"""Ретрай задачи после успешной вставки: где ClickHouse отбросит повтор, а где задвоит строки."""
from datetime import datetime, timedelta

from airflow.providers.clickhousedb.hooks.clickhouse import ClickHouseHook
from airflow.sdk import dag, get_current_context, task

ROWS = 1000
COLUMNS = ["id", "event", "loaded_at"]
WINDOW = "SETTINGS non_replicated_deduplication_window = 1000"

SYNC = {"async_insert": 0}  # в ClickHouse 26.9 async_insert включён по умолчанию

# таблица -> (настройки MergeTree, токен: False / True / "per_batch", доп. настройки сессии, batch_size)
CASES = {
    "events_no_window": ("", True, SYNC, None),                 # токен есть, окна нет
    "events_no_token": (WINDOW, False, SYNC, None),             # окно есть, токена нет
    "events_token_sync": (WINDOW, True, SYNC, None),            # окно и токен, синхронная вставка
    "events_token_batches": (WINDOW, True, SYNC, 500),          # один токен на все пачки
    "events_token_per_batch": (WINDOW, "per_batch", SYNC, 500), # свой токен у каждой пачки
    "events_token_async": (WINDOW, True, {}, None),             # окно и токен, дефолты сервера
    "events_token_async_dedup": (WINDOW, True, {"async_insert_deduplicate": 1}, None),
}


def make_rows():
    # loaded_at меняется от попытки к попытке, поэтому хеш блока у повтора другой
    loaded_at = datetime.now()
    return [(i, f"click_{i % 5}", loaded_at) for i in range(ROWS)]


@dag(schedule=None, start_date=datetime(2026, 10, 1), catchup=False, tags=["clickhouse"])
def retry_dedup():
    @task
    def prepare():
        hook = ClickHouseHook()
        for table, (mt_settings, *_rest) in CASES.items():
            hook.run(
                f"CREATE OR REPLACE TABLE {table} "
                f"(id UInt32, event String, loaded_at DateTime64(6)) "
                f"ENGINE = MergeTree ORDER BY id {mt_settings}"
            )

    @task(retries=1, retry_delay=timedelta(seconds=5))
    def load(table: str):
        _, use_token, extra, batch_size = CASES[table]
        ctx = get_current_context()
        ti = ctx["ti"]
        settings = dict(extra)
        # токен детерминирован: одинаков у первой попытки и у ретрая
        token = f"{ti.dag_id}:{ctx['run_id']}:{ti.task_id}"
        rows = make_rows()
        if use_token == "per_batch":
            # bulk_insert_rows не умеет менять настройки между пачками, поэтому клиент напрямую
            hook = ClickHouseHook(session_settings=settings)
            client = hook.get_client()
            try:
                for n, start in enumerate(range(0, len(rows), batch_size)):
                    client.insert(
                        table,
                        rows[start : start + batch_size],
                        column_names=COLUMNS,
                        settings={"insert_deduplication_token": f"{token}:batch{n}"},
                    )
            finally:
                client.close()
        else:
            if use_token:
                settings["insert_deduplication_token"] = token
            # параметра settings у bulk_insert_rows нет, токен едет через session_settings хука
            hook = ClickHouseHook(session_settings=settings)
            hook.bulk_insert_rows(table, rows, COLUMNS, batch_size=batch_size)
        if ti.try_number == 1:
            raise RuntimeError("сбой после вставки: строки уже в ClickHouse, задача уходит в ретрай")

    @task(trigger_rule="all_done")
    def report():
        hook = ClickHouseHook()
        for table in CASES:
            rows = hook.get_first(f"SELECT count() FROM {table}")[0]
            print(f"{table}: {rows} строк при ожидаемых {ROWS}")

    loads = [load.override(task_id=f"load_{table}")(table) for table in CASES]
    prepare() >> loads >> report()


retry_dedup()

Запускаем DAG командой airflow dags test. В Airflow 3.3.2 она честно ждёт retry_delay и перезапускает упавшие задачи, отдельный планировщик не нужен. Из лога берём число ретраев, строки задачи report и итог запуска.

# Airflow 3.3.2, прогнано на стенде 2026-10-21
docker compose exec -T airflow bash -c "airflow dags test retry_dedup > /tmp/dagtest.log 2>&1; rc=\$?; echo \"задач ушло в ретрай: \$(grep -c 'new_state=up_for_retry' /tmp/dagtest.log)\"; grep -E 'строк при' /tmp/dagtest.log; grep -oE 'run_duration=[0-9.]+, state=[a-z]+' /tmp/dagtest.log; exit \$rc"

задач ушло в ретрай: 7
events_no_window: 2000 строк при ожидаемых 1000
events_no_token: 2000 строк при ожидаемых 1000
events_token_sync: 1000 строк при ожидаемых 1000
events_token_batches: 500 строк при ожидаемых 1000
events_token_per_batch: 1000 строк при ожидаемых 1000
events_token_async: 1000 строк при ожидаемых 1000
events_token_async_dedup: 1000 строк при ожидаемых 1000
run_duration=19.847764, state=success

Разберём таблицы по порядку, потому что каждая отвечает на свой вопрос.

• events_no_window. Токен есть, но у MergeTree окно равно нулю. Сервер не хранит лог дедупликации, и ретрай удваивает данные.
• events_no_token. Окно есть, токена нет. Из-за loaded_at хеш повтора отличается, и строки снова задвоены.
• events_token_sync. Окно и токен вместе дают ровно 1000 строк. Это рабочий рецепт.
• events_token_batches. Хук отправил две пачки по 500 строк с одним токеном из session_settings. Вторую пачку сервер отбросил как дубль первой ещё на первой попытке, ретрай тоже отброшен, и в таблице осталась половина данных.
• events_token_per_batch. Свой токен у каждой пачки через get_client() возвращает правильные 1000 строк.
• events_token_async и events_token_async_dedup. С асинхронной вставкой токен тоже работает, причём значение async_insert_deduplicate на результат не повлияло.

Таблица events_token_batches это главный практический вывод статьи. Потеря данных здесь не связана с ретраем вообще, она случается на первой же попытке, стоит только передать токен через хук и задать batch_size.

Чтобы убедиться, что дубли отбросил именно сервер, а не клиент, смотрим в query_log. Скрипт считает вставки по каждой таблице, их асинхронную часть и число строк, которые клиент отправил.

-- ClickHouse 26.9.23.15, прогнано на стенде 2026-10-21
-- Что происходило со вставками DAG retry_dedup: async или нет, сколько блоков отброшено как дубли
SYSTEM FLUSH LOGS;
SELECT
    tables[1] AS table_name,
    count() AS inserts,
    sum(ProfileEvents['AsyncInsertQuery']) AS async_inserts,
    sum(written_rows) AS sent_rows
FROM system.query_log
WHERE type = 'QueryFinish' AND query_kind = 'Insert' AND table_name LIKE 'demo.events%'
GROUP BY table_name
ORDER BY table_name
FORMAT PrettyCompactMonoBlock;
-- дубли считаются там, где блок реально пишется: в самом INSERT или в сбросе async-буфера
SELECT
    sum(ProfileEvents['DuplicatedInsertedBlocks']) AS duplicated_blocks_total
FROM system.query_log
WHERE type = 'QueryFinish' AND has(tables, 'demo.events_token_async')
FORMAT PrettyCompactMonoBlock;

# ClickHouse 26.9.23.15, прогнано на стенде 2026-10-21
docker compose exec -T clickhouse clickhouse-client --user airflow --password airflow --multiquery < dedup_stats.sql

   ┌─table_name────────────────────┬─inserts─┬─async_inserts─┬─sent_rows─┐
1. │ demo.events_no_token          │       2 │             0 │      2000 │
2. │ demo.events_no_window         │       2 │             0 │      2000 │
3. │ demo.events_token_async       │       2 │             2 │      2000 │
4. │ demo.events_token_async_dedup │       2 │             2 │      2000 │
5. │ demo.events_token_batches     │       4 │             0 │      2000 │
6. │ demo.events_token_per_batch   │       4 │             0 │      2000 │
7. │ demo.events_token_sync        │       2 │             0 │      2000 │
   └───────────────────────────────┴─────────┴───────────────┴───────────┘
   ┌─duplicated_blocks_total─┐
1. │                       1 │
   └─────────────────────────┘

Клиент отправил по 2000 строк в каждую таблицу, две попытки по 1000. У таблиц с пачками по четыре вставки, у асинхронных вариантов все вставки прошли через буфер. Счётчик DuplicatedInsertedBlocks для events_token_async равен единице, то есть повторный блок отбросил сервер при сбросе буфера.

 

Замер трёх путей вставки

Последний скрипт сравнивает пути вставки на таблице из пяти колонок. Пакетные пути получают 1 млн строк и прогоняются три раза, в отчёт идёт медиана. Построчный insert_rows из common-sql, тот же путь DB-API, которым пользуются универсальные SQL-операторы, получает 2000 строк и один прогон, потому что миллион строк он грузил бы часами. После каждого прогона скрипт проверяет число строк в таблице.

# apache-airflow-providers-clickhousedb 1.0.1, airflow-clickhouse-plugin 1.8.0, clickhouse-connect 1.11.0, ClickHouse 26.9.23.15, прогнано на стенде 2026-10-21
# Замер путей вставки. Запуск внутри контейнера Airflow: соединения берутся из AIRFLOW_CONN_*
import statistics
import time
from datetime import datetime, timedelta

from airflow.providers.clickhousedb.hooks.clickhouse import ClickHouseHook
from airflow_clickhouse_plugin.hooks.clickhouse import ClickHouseHook as PluginHook

N = 1_000_000          # строк для пакетных путей
N_ROWWISE = 2_000      # построчный insert_rows на миллионе не дождаться
REPEATS = 3
COLUMNS = ["id", "ts", "user_id", "event", "amount"]
TABLE = "bench_events"

base = datetime(2026, 10, 1)
ROWS = [(i, base + timedelta(seconds=i), i % 50_000, f"event_{i % 20}", i * 0.01) for i in range(N)]

admin = ClickHouseHook()
admin.run(
    f"CREATE OR REPLACE TABLE {TABLE} (id UInt64, ts DateTime, user_id UInt32, "
    f"event LowCardinality(String), amount Float64) ENGINE = MergeTree ORDER BY (event, ts)"
)


def bulk_default(rows):
    # дефолты сервера 26.9: async_insert=1, wait_for_async_insert=1
    ClickHouseHook().bulk_insert_rows(TABLE, rows, COLUMNS)


def bulk_sync(rows):
    ClickHouseHook(session_settings={"async_insert": 0}).bulk_insert_rows(TABLE, rows, COLUMNS)


def bulk_sync_batches(rows):
    # один insert context на все пачки, память ограничена размером пачки
    ClickHouseHook(session_settings={"async_insert": 0}).bulk_insert_rows(TABLE, rows, COLUMNS, batch_size=100_000)


def plugin_native(rows):
    # clickhouse-driver, нативный протокол 9000, settings передаются прямо в execute
    PluginHook(clickhouse_conn_id="clickhouse_native").execute(
        f"INSERT INTO {TABLE} VALUES", rows, settings={"async_insert": 0}
    )


def dbapi_rowwise(settings, executemany=False):
    # общий insert_rows из common-sql: курсор DB-API, INSERT на каждую строку
    def run(rows):
        ClickHouseHook(session_settings=settings).insert_rows(
            TABLE, rows, COLUMNS, commit_every=1000, executemany=executemany
        )
    return run


def measure(name, fn, rows, repeats):
    times = []
    for _ in range(repeats):
        admin.run(f"TRUNCATE TABLE {TABLE}")
        t = time.perf_counter()
        fn(rows)
        times.append(time.perf_counter() - t)
        got = admin.get_first(f"SELECT count() FROM {TABLE}")[0]
        assert got == len(rows), f"{name}: {got} строк вместо {len(rows)}"
    med = statistics.median(times)
    print(f"{name:<40} {len(rows):>9} строк  медиана {med:8.2f} с  {len(rows) / med:>12,.0f} строк/с")


print(f"{'путь':<40} {'объём':>9}        {'время':>14}  {'скорость':>15}")
measure("bulk_insert_rows, дефолты сервера", bulk_default, ROWS, REPEATS)
measure("bulk_insert_rows, async_insert=0", bulk_sync, ROWS, REPEATS)
measure("bulk_insert_rows, batch_size=100000", bulk_sync_batches, ROWS, REPEATS)
measure("airflow-clickhouse-plugin, native", plugin_native, ROWS, REPEATS)
measure("insert_rows (DB-API), дефолты сервера", dbapi_rowwise({}), ROWS[:N_ROWWISE], 1)
measure("insert_rows (DB-API), async_insert=0", dbapi_rowwise({"async_insert": 0}), ROWS[:N_ROWWISE], 1)
measure("insert_rows executemany, async_insert=0", dbapi_rowwise({"async_insert": 0}, True), ROWS[:N_ROWWISE], 1)

# Docker Compose v2, прогнано на стенде 2026-10-21
docker compose exec -T -e AIRFLOW__LOGGING__LOGGING_LEVEL=WARNING airflow sh -c "cd /opt/airflow/demo && python bench_insert.py"

2026-10-21T15:58:28.251474Z [warning  ] The `airflow.hooks.base.BaseHook` attribute is deprecated. Please use `'airflow.sdk.bases.hook.BaseHook'`. [py.warnings] category=DeprecatedImportWarning filename=/home/airflow/.local/lib/python3.12/site-packages/airflow_clickhouse_plugin/hooks/clickhouse.py lineno=6
путь                                         объём                 время         скорость
bulk_insert_rows, дефолты сервера          1000000 строк  медиана     1.31 с       760,495 строк/с
bulk_insert_rows, async_insert=0           1000000 строк  медиана     1.24 с       806,687 строк/с
bulk_insert_rows, batch_size=100000        1000000 строк  медиана     1.36 с       735,734 строк/с
airflow-clickhouse-plugin, native          1000000 строк  медиана     3.97 с       251,615 строк/с
insert_rows (DB-API), дефолты сервера         2000 строк  медиана   123.52 с            16 строк/с
insert_rows (DB-API), async_insert=0          2000 строк  медиана    15.48 с           129 строк/с
insert_rows executemany, async_insert=0       2000 строк  медиана    13.85 с           144 строк/с

Пакетная вставка провайдера держит 736-807 тысяч строк в секунду, и выключение async_insert на пакете в миллион строк почти ничего не меняет. Разница 1.31 и 1.24 секунды лежит в пределах шума. Плагин на том же объёме втрое медленнее. Самое интересное в нижней части таблицы. Построчный insert_rows при дефолтах сервера 26.9 даёт 16 строк в секунду, потому что каждая строка становится отдельной асинхронной вставкой и ждёт сброса буфера. Выключение async_insert ускоряет его в восемь раз, но 129 строк в секунду всё равно почти на четыре порядка хуже пакетного пути, а executemany строки не группирует. Предупреждение DeprecatedImportWarning в начале вывода печатает плагин, он импортирует BaseHook по пути из Airflow 2.

 

Ограничения и подводные камни

Прогон показал несколько мест, где поведение провайдера и сервера расходится с интуицией. Их стоит держать в голове при переносе DAG на новый пакет.

• Общий токен на пачки. bulk_insert_rows с batch_size и токеном в session_settings молча теряет все пачки, кроме первой. Ошибки нет, в логе хука написано, что строки вставлены.
• Нулевое окно у MergeTree. Токен без non_replicated_deduplication_window на нереплицируемой таблице ничего не делает.
• Дефолты 26.x для построчных вставок. При async_insert и wait_for_async_insert, включённых по умолчанию, вставки через курсор DB-API становятся очень медленными.
• Границы окна. Если между первой попыткой и ретраем в таблицу прошло больше вставок, чем вмещает окно, дубль пройдёт, это прямо оговорено в документации ClickHouse.

Кроме того, токен защищает только от повтора той же самой вставки. Если задача между попытками формирует другой набор строк, например читает источник с плавающей границей, токен отбросит новые данные вместо старых, и это уже ошибка в логике DAG.

 

Когда переходить на официальный провайдер

Для новых DAG на Airflow 2.11 и новее провайдер clickhousedb стоит брать по умолчанию. Он сопровождается рядом с самим Airflow, работает по HTTP, который проще пропустить через прокси и балансировщики, а пакетная вставка у него самая быстрая из проверенных. Загрузку при этом лучше строить на bulk_insert_rows без batch_size, когда объём помещается в память воркера, и с токеном из идентификаторов DAG, запуска и задачи.

Если же данные нужно резать на пачки или передавать разные настройки разным вставкам, пока удобнее get_client() с явным токеном на каждую пачку. Существующие DAG на airflow-clickhouse-plugin переписывать срочно незачем, пакет работает на Airflow 3.3.2. Миграция окупается, когда объёмы растут и плагин начинает отставать по скорости, или когда предупреждения об устаревших импортах превращаются в ошибки. Оба пакета живут в разных пространствах имён и могут стоять в одном окружении, так что переносить DAG можно по одному. Системно оркестрацию, хуки и надёжные конвейеры разбирают на нашем курсе по построению конвейеров данных в Apache Airflow.

 

Заключение

Официальный провайдер ClickHouse для Airflow снимает вопрос «чем подключаться», но не вопрос «как не задвоить данные». Ретрай задачи остаётся доставкой at-least-once, и защищает от неё только связка из ненулевого окна дедупликации и детерминированного insert_deduplication_token. На стенде эта связка отработала и в синхронном, и в асинхронном режиме. Единственная ловушка сидит в самом провайдере, потому что пачки bulk_insert_rows разделяют один токен сессии, и пока это не исправлено, токен на каждую пачку придётся задавать через сырой клиент.

 

Референсные ссылки

• Анонс провайдера ClickHouse для Apache Airflow в блоге ClickHouse
• Документация apache-airflow-providers-clickhousedb
• Дедупликация вставок при ретраях в документации ClickHouse
• Асинхронные вставки в документации ClickHouse
• Пакет airflow-clickhouse-plugin на PyPI