Apache Airflow 向けの公式 ClickHouse プロバイダーが利用可能に

はじめに #

多くの ClickHouse ユーザーは、データパイプラインをオーケストレーションしてデータの取り込みや変換、定期的な分析ジョブをスケジュールするオープンソースの標準ツールとして、Apache Airflow を活用しています。これまで両者を接続するには、コミュニティプラグインをインストールするか、独自の統合コードを作成するのが一般的でした。

このたび、Airflow のアップストリームに公式の ClickHouse プロバイダー apache-airflow-providers-clickhousedb が追加されました。ClickHouse Connect を介して HTTP(S) 経由で接続し、Airflow の共通 SQL オペレーターに対応するほか、一括処理やクライアント固有の操作を行うための ClickHouse フックも備えています。本記事では、このプロバイダーのインストール方法、接続の設定手順、セルフマネージドの Airflow 環境や Astronomer などのマネージドプラットフォーム上で同じワークフローを実行する方法について解説します。

コミュニティでの発足の経緯 #

今回のリリース以前は、ClickHouse コミュニティが独自にこの課題を解決していました。Airflow にネイティブな ClickHouse 連携機能がなかった当時、Anton Bryzgalov 氏(bryzgaloff)が airflow-clickhouse-plugin を開発しました。同氏が長年にわたりメンテナンスを続けた結果、このプラグインは Airflow および ClickHouse コミュニティにおけるデファクトスタンダードとなり、PyPI でもダウンロード数上位 1% に入るパッケージへと成長しました。ClickHouse 社内のデータウェアハウスチームが構築した社内ツールの設計にも、このプラグインの慣例が反映されています。ClickHouse のエコシステムが現在のように充実しているのは、こうした貢献のおかげです。Anton 氏に深く感謝いたします。

公式にメンテナンスされた統合環境を求めるチームにとって、このプロバイダーへの移行は自然なアップグレードパスとなります。今後のリソース投下や新機能の追加はこちらを中心に行われますが、移行作業の大半は機械的な手順で完了します。プロバイダーをインストールし、接続先を HTTP(S) ポートに向けたうえで、DAG 内で標準の SQLExecuteQueryOperator を使用するだけです。

公式プロバイダーを提供する理由 #

ClickHouse では絶えず新機能をリリースしています。アップストリームに公式プロバイダーが存在することで、外部の追従を待つことなく、データベース本体の進化に足並みを揃えて統合機能を提供できるようになります。

今回の実装にあたっては、以下のような設計方針を採用しました。

• ClickHouse Connect 上に構築: プロバイダーは、ClickHouse 社内でメンテナンスしている Python クライアント clickhouse-connect を使い、HTTP インターフェース経由で接続します。クライアントの高速化や機能追加が行われると、プロバイダーにもそのまま反映されます。
• Airflow の共通 SQL フレームワークを採用: プロバイダーは apache-airflow-providers-common-sql 経由で ClickHouse を公開するため、標準の SQLExecuteQueryOperator で DDL、DML、分析クエリを処理できます。ClickHouse 固有のオペレーターを新たに覚える必要はありません。
• それ以外の処理には専用フックを用意: 一括挿入やストリーミング、ClickHouse 固有のクライアント呼び出しには、ClickHouseHook を介して直接アクセスできます。ネイティブな列指向の挿入パスを利用する bulk_insert_rows メソッドも用意されています。

Airflow と ClickHouse の活用事例 #

現在、多くのユーザーが ClickHouse と Airflow を組み合わせて運用しています。この組み合わせは、ClickHouse がサービスを提供するほぼすべての業界に加え、ClickHouse 自身の社内スタックでも活用されています。

Astronomer との関係も双方向のものです。同社のデータオブザーバビリティ製品である Astro Observe は ClickHouse Cloud 上に構築されており、数十億件にのぼる Airflow ワークフローイベントを処理して、Airflow ユーザーにパイプラインのリアルタイムなインサイトを提供しています。数千社におよぶ企業に向けて Airflow 運用プラットフォームを提供するチーム自身が、自社の分析基盤として ClickHouse を選択しました。

ストリーミングプラットフォームやソーシャルメディア上で 1,200 万組以上のアーティストの動向を追跡する Chartmetric は、Airflow でオーケストレーションされたパイプラインと ClickHouse Cloud を組み合わせており、その中には 5 分ごとに 1,500 万行以上を取り込むプレイリストキャッシュパイプラインも含まれています。

ClickHouse 社内でもまったく同じパターンを採用しています。社内データウェアハウスは ClickHouse Cloud 上に構築されており、Airflow が 40 以上のデータソースにわたる 76 の DAG で挿入ジョブをスケジュールし、1 日あたり約 60 億行のデータを処理しています。週次メトリクスを確認する経営層から、日々の問い合わせに対応するプロダクト、営業、サポートの各チーム、さらには自社データ上で構築を進めているエージェント型ワークフローに至るまで、全社がこの基盤に依存しています。Airflow は、それらすべてにデータを供給し続ける中核コンポーネントとなっています。

Apache Airflow での始め方 #

オープンソースの Airflow を実行している場合、他のプロバイダーと同様にインストールできます。

1pip install apache-airflow-providers-clickhousedb

これにより、apache-airflow-providers-common-sql と clickhouse-connect が自動的に導入されます。次に、接続を作成します。プロバイダーによって clickhouse 接続タイプが登録されるため、Airflow UI の Admin > Connections から設定するか、環境変数として定義できます。

1export AIRFLOW_CONN_CLICKHOUSE_DEFAULT='{
2    "conn_type": "clickhouse",
3    "host": "abc123.clickhouse.cloud",
4    "port": 8443,
5    "login": "default",
6    "password": "secret",
7    "schema": "my_database",
8    "extra": {"secure": true}
9}'

ClickHouse Cloud や TLS が有効なクラスターの場合は、secure を true に設定し、ポート 8443 を指定します。

接続が完了すれば、DAG 自体は標準的な Airflow のコードと変わりません。

1from datetime import datetime
2
3from airflow import DAG
4from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator
5
6with DAG(
7    dag_id="clickhouse_example",
8    start_date=datetime(2026, 1, 1),
9    default_args={"conn_id": "clickhouse_default"},
10    schedule="@daily",
11    catchup=False,
12) as dag:
13    create_table = SQLExecuteQueryOperator(
14        task_id="create_table",
15        sql="""
16            CREATE TABLE IF NOT EXISTS events_daily (
17                day  Date,
18                user_id String,
19                events UInt64
20            ) ENGINE = MergeTree()
21            ORDER BY (day, user_id);
22        """,
23    )
24
25    aggregate = SQLExecuteQueryOperator(
26        task_id="aggregate_events",
27        sql="""
28            INSERT INTO events_daily
29            SELECT toDate(ts), user_id, count()
30            FROM events
31            WHERE toDate(ts) = yesterday()
32            GROUP BY toDate(ts), user_id;
33        """,
34    )
35
36    create_table >> aggregate

SQL オペレーターでは対応できないワークロードの場合、ClickHouseHook を使って基盤となるクライアントに直接アクセスできます。

1from airflow.providers.clickhousedb.hooks.clickhouse import ClickHouseHook
2
3hook = ClickHouseHook(clickhouse_conn_id="clickhouse_default")
4hook.bulk_insert_rows(
5    table="events",
6    rows=[("user1", "click"), ("user2", "view")],
7    column_names=["user_id", "action"],
8    batch_size=1000,
9)

セッション設定、タスク単位でのデータベースオーバーライド、接続オプションを含む詳しいチュートリアルは、ドキュメントに記載されています。実際の動作を確認したい場合は、2026年5月に開催されたユーザーカンファレンス Open House 2026 のエコシステムセッション内で、Bentsi Leviav によるプロバイダーのデモをご覧ください。

Astronomer での始め方 #

Astronomer は、多くのユーザーが本番環境で運用しているマネージド Airflow プラットフォームであり、Astro CLI はローカルに Airflow 環境を立ち上げる最も手軽な方法です。今回のプロバイダーは特別な設定なしでそのまま動作します。

まず、CLI をインストールしてプロジェクトの雛形を作成します。

1brew install astro
2astro dev init

作成したプロジェクト内の requirements.txt にプロバイダーを追加します。

1apache-airflow-providers-clickhousedb

続いて、Airflow をローカルで起動します。

1astro dev start

これで、手元のマシン上でコンテナとして Airflow のコンポーネントが立ち上がります。起動したら、ブラウザで localhost:8080 を開いて Airflow UI にアクセスし、Admin > Connections から ClickHouse タイプの接続を作成して、ClickHouse Cloud サービスまたはセルフホストのクラスターを指定します(TLS 接続用に secure: true とポート 8443 の設定をお忘れなく)。

前述のセクションの DAG を dags/ フォルダに配置すると UI 上に表示され、すぐにトリガーできるようになります。

Astro を利用している場合は、さらに簡単な接続方法があります。Astro UI の Environment Manager を使うと、ClickHouse 接続を一度作成するだけで、認証情報を Astro のマネージドシークレットバックエンドに保存し、ワークスペース内の全デプロイメントで共有できます。必要に応じてデプロイメントごとにオーバーライドすることも可能です。Astro CLI はそれらの接続情報をローカル環境にプルできるため、一度 ClickHouse を設定すればローカルでもホスト環境でも共通して使用できます。

本番環境へのリリースの準備が整ったら、astro deploy を実行するだけで、プロバイダーを含めた同一のプロジェクトが Astro のデプロイメントに反映されます。ローカルと本番環境の間で ClickHouse の設定を変更する必要はありません。

今後の展望 #

このプロバイダーは本日より利用可能であり、アーリーアダプターの手によってすでに本番環境で大規模に運用されています。今後の新機能の開発はコミュニティからの要望に基づいて優先順位を決定しますので、必要な機能があれば Issue や PR を通してお知らせください。

現在 Airflow で ClickHouse を運用している方は、ぜひ使い心地をお聞かせください。ClickHouse Community Slack でのご参加をお待ちしています。また、ClickHouse を初めて利用される場合は、300 ドル分の無料クレジットを活用して、数分で ClickHouse Cloud を試すことができます。皆様がどのようなシステムを構築されるか楽しみにしています。