ClickHouse 收編官方 Airflow provider，SQL operator 取代自訂 hook
ClickHouse Blog · 2026-09-17
ClickHouse 把過去由社群維護的 Airflow 外掛收編成官方套件，讓原本要自己寫 PythonOperator 包 clickhouse-connect 呼叫的 ETL 任務，換成標準的 SQLExecuteQueryOperator 與 ClickHouseHook。官方套件 apache-airflow-providers-clickhousedb 1.0.0 已於 2026 年 6 月 22 日上架 PyPI，ClickHouse 官方部落格則在 2026 年 9 月 17 日發文正式宣布。
背景
在官方套件出現前，團隊要在 Airflow 排程寫入 ClickHouse，大多靠社群套件或自己寫 PythonOperator 包一層 hook。ClickHouse 官方特別點名 Anton Bryzgalov 維護的 airflow-clickhouse-plugin，形容它是「PyPI 上下載量前 1% 的套件」，代表這條路早已被大量團隊踩過。
問題在於這條路沒有官方背書：連線設定、批次寫入、重試邏輯全部要團隊自己維護，每個 DAG 重複實作一次 hook，版本升級時也沒有隨 Airflow 生態系統一起走的相容性保證。
核心改動
新套件把功能拆成兩層。DDL、DML 與一般查詢交給 Airflow 內建的 SQLExecuteQueryOperator（來自 apache-airflow-providers-common-sql），批次寫入與 ClickHouse 專屬操作則由新增的 ClickHouseHook 負責，內建 bulk_insert_rows 方法。連線類型註冊為 clickhouse，底層走 ClickHouse Connect 的 HTTP(S) 介面。
依賴版本明確寫在套件中：apache-airflow>=2.11.0、apache-airflow-providers-common-sql>=1.32.0、clickhouse-connect>=1.3.0，這三個版本號同時出現在 Airflow 官方文件與 PyPI 頁面。
過去常見的寫法，是在 PythonOperator 裡手動建立 client、手動組 insert：
def load_to_clickhouse(**context):
import clickhouse_connect
client = clickhouse_connect.get_client(
host="abc123.clickhouse.cloud", port=8443,
username="default", password=Variable.get("ch_pwd"))
client.insert("events", rows, column_names=["user_id", "action"])
load = PythonOperator(task_id="load", python_callable=load_to_clickhouse)換成新套件後，連線設定移到 Airflow Connection，DAG 裡只留業務邏輯：
from airflow.providers.clickhousedb.hooks.clickhouse import ClickHouseHook
hook = ClickHouseHook(clickhouse_conn_id="clickhouse_default")
hook.bulk_insert_rows(
table="events",
rows=[("user1", "click"), ("user2", "view")],
column_names=["user_id", "action"],
batch_size=1000,
)DDL 類任務則直接用 SQLExecuteQueryOperator，不再需要自己包 hook：
create_table = SQLExecuteQueryOperator(
task_id="create_table",
sql="""
CREATE TABLE IF NOT EXISTS events_daily (
day Date, user_id String, events UInt64
) ENGINE = MergeTree() ORDER BY (day, user_id);
""",
)連線資訊也可以用環境變數宣告，ClickHouse Cloud 走 TLS 需指定連接埠 8443 並加 secure: true：
export AIRFLOW_CONN_CLICKHOUSE_DEFAULT='{
"conn_type": "clickhouse", "host": "abc123.clickhouse.cloud",
"port": 8443, "login": "default", "password": "secret",
"schema": "my_database", "extra": {"secure": true}
}'影響範圍
現有用 airflow-clickhouse-plugin 或自寫 hook 的 DAG 是首要遷移對象：批次寫入邏輯可以直接換成 ClickHouseHook.bulk_insert_rows，查詢類任務改成 SQLExecuteQueryOperator，連線設定搬進 Airflow 的 Connections 介面或環境變數，不用再把密碼寫死在 DAG 檔案裡。
要升級的團隊得先確認執行環境滿足 apache-airflow>=2.11.0 與 apache-airflow-providers-common-sql>=1.32.0，否則 pip install apache-airflow-providers-clickhousedb 會因依賴衝突失敗；用 Astronomer 的團隊則是把套件加進 requirements.txt 後跑 astro dev start 重建映像。
這個整合已經不是 demo 等級。ClickHouse 官方部落格提到，內部資料倉儲用同一套 provider 每天處理約 60 億列、跑在 76 個 DAG 上，客戶 Chartmetric 也在生產環境使用，算是釋出前就先自己吃了狗糧。