> ## 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. > コミュニティ版 airflow-clickhouse-plugin から公式の apache-airflow-providers-clickhousedb プロバイダへ、既存の Airflow DAG を移行します # airflow-clickhouse-plugin から ClickHouse プロバイダへの移行 [公式の ClickHouse プロバイダ](/ja/integrations/connectors/data-ingestion/etl-tools/airflow-and-clickhouse)が登場する以前、ほとんどの Airflow ユーザーはコミュニティパッケージ [airflow-clickhouse-plugin](https://github.com/bryzgaloff/airflow-clickhouse-plugin) を介して ClickHouse に接続していました。本ガイドでは、既存のデプロイメントを `apache-airflow-providers-clickhousedb` へ移行する手順を説明します。プロバイダ自体の動作については、[Apache Airflow と ClickHouse の接続](/ja/integrations/connectors/data-ingestion/etl-tools/airflow-and-clickhouse)を参照してください。

なぜネイティブプロバイダなのか？

プロバイダは、Airflow がサードパーティ製システムと連携するための仕組みです。Airflow エコシステムの他の構成要素と同じタイミングでリリース・テスト・ドキュメント化されており、本プロバイダは Airflow コミュニティと ClickHouse チームが共同でメンテナンスしています。コミュニティがメンテナンスするドライバーではなく、ClickHouse 自身が開発・サポートする Python クライアントである [clickhouse-connect](/ja/integrations/language-clients/python/index) を基盤としています。そのため、新しいサーバー機能や修正は、サポートされた経路を通じて Airflow ユーザーのもとに届きます。プロバイダに移行すれば、公式の提供元を持つパッケージ、標準の `common.sql` オペレーターとセンサー、そして他のデータベースと同様に Airflow UI に表示される接続タイプが利用できます。 2 つのパッケージの違いは import パスだけではありません。プラグインは `clickhouse-driver` を用いて **ネイティブ TCP プロトコル** で ClickHouse と通信します。一方プロバイダは `clickhouse-connect` を用いて **HTTP(S)** で通信し、ClickHouse 固有のオペレーターを同梱する代わりに、汎用の `common.sql` オペレーターに組み込まれます。 変更を加える前に、このガイドを一度通読してください。特に接続の変更は、すべての DAG に同時に影響します。

概要

| | `airflow-clickhouse-plugin` | `apache-airflow-providers-clickhousedb` | | - | - | - | | インポートルート | `airflow_clickhouse_plugin` | `airflow.providers.clickhousedb` | | ドライバー | `clickhouse-driver` | `clickhouse-connect` | | プロトコル / デフォルトポート | ネイティブ TCP、`9000` (TLS 使用時は `9440`) | HTTP、`8123` (TLS 使用時は `8443`) | | 接続タイプ | 登録なし。任意のタイプで動作 | `clickhouse` | | 接続の extras | `clickhouse_driver.Client` にそのまま渡される | 固定のオプションセット。[追加の接続オプション](/ja/integrations/connectors/data-ingestion/etl-tools/airflow-and-clickhouse#extra-connection-options)を参照 | | オペレーターとセンサー | `ClickHouseOperator`、`ClickHouseSensor`、および `ClickHouse` プレフィックス付きの `common.sql` ラッパー | `common.sql` のオペレーターとセンサーをそのまま使用 | | フック | `ClickHouseHook` (`BaseHook`) と `ClickHouseDbApiHook` (`DbApiHook`) | `ClickHouseHook` (`DbApiHook`) の 1 つのみ | | 最小 Airflow バージョン | 2.0 | 2.11 |

ステップ 1: 前提条件の確認とインストール

このプロバイダには Airflow 2.11 以降と `apache-airflow-providers-common-sql` 1.32.0 以降が必要です。これより古いバージョンを使用している場合は、まず Airflow をアップグレードしてください。 ```bash theme={null} pip install apache-airflow-providers-clickhousedb ``` 2つのパッケージは異なるPythonネームスペースに存在するため、DAGを1つずつ移行しながら両方を並行してインストールしておけます。どこからもインポートされなくなったら、プラグインを削除してください: ```bash theme={null} pip uninstall airflow-clickhouse-plugin clickhouse-driver ```

ステップ 2: connection を更新する

このステップを省略すると動作しなくなります。既存の ClickHouse connection はすべてネイティブポートを指しているため、provider が必要とする HTTP ポートに変更する必要があります。 | フィールド | Plugin | Provider | | - | - | - | | 接続タイプ | 任意 (多くの場合 `sqlite` または `generic`) | `clickhouse` | | ポート (平文) | `9000` | `8123` | | ポート (TLS) | `9440` | `8443` | | Login | ドライバーのデフォルトである `default` | 同上 | | スキーマ | データベース | 同上 | ClickHouse がファイアウォールや load balancer の背後にある場合は、切り替える前に worker から HTTP ポートに到達できることを確認してください。あわせて、server 側で HTTP インターフェイスが有効になっているか (server configuration の `http_port` または `https_port`) も確認してください。[ClickHouse Cloud](/ja/products/cloud/getting-started/intro) では HTTPS は `8443` のみで公開されています。 URI として保存されている connection (`clickhouse://user:pass@host:9000/db?secure=true`) の場合、Airflow が URI のスキームから判別するため、接続タイプはすでに `clickhouse` になっています。変更が必要なのはポートと extra のオプションだけです。クエリ文字列の値は JSON としてパースされるため、`secure=true` はブール値のまま扱われます。

接続の extra

プラグインは `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/` を送信し、指定した値をラベルとして末尾に追加します。 | | `ca_certs` | `client_kwargs.ca_cert` | CA bundle へのパス。 | | `certfile` / `keyfile` | `client_kwargs.client_cert` / `client_kwargs.client_cert_key` | mutual TLS。 | | `server_hostname` | `client_kwargs.server_host_name` | TLS の SNI オーバーライド。 | | `alt_hosts`, `round_robin` | 同等の設定なし | クライアントは接続に指定された単一のホストにのみ接続します。フェイルオーバーリストを利用していた場合は、接続先をロードバランサーまたは 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 を変更する前に、移行した各 connection を検証します。 ```bash theme={null} airflow connections test clickhouse_default ```

ステップ3: import の置き換え

| プラグインのクラス | プロバイダでの置き換え先 | | - | - | | `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` | `ClickHouse` プレフィックスが付いた `common.sql` の wrapper は、プラグインの hook を注入するためだけに存在していたものです。プロバイダが `clickhouse` 接続タイプを登録するため、プレフィックスなしの `common.sql` のクラスは connection から hook を自力で解決できます。これらの wrapper を使用していた場合、移行作業は通常、import 行の書き換え、`ClickHouse` プレフィックスの削除、`conn_id` の明示的な指定だけで済みます ([ステップ7](#step-7-the-commonsql-wrapper-family)を参照) 。

ステップ4: `ClickHouseOperator` から `SQLExecuteQueryOperator` へ

| `ClickHouseOperator` の argument | `SQLExecuteQueryOperator` での同等な指定 | Notes | | - | - | - | | `sql` | `sql` | 文字列、文字列のリスト、または `.sql` ファイルの file path。どちらもテンプレート化に対応しています。 | | `clickhouse_conn_id` | `conn_id` | plugin のデフォルトは `clickhouse_default` でした。`SQLExecuteQueryOperator` にデフォルトはないため、task ごとに `conn_id` を渡すか、`default_args` で一度だけ設定してください。 | | `database` | `database` | 変更ありません。 | | `SELECT` 向けの `parameters` | `parameters` | `%(name)s` の placeholder は引き続き使用できます。`{name:Type}` によるサーバー側の binding も利用可能です。 | | `INSERT` 向けの `parameters` (行のリスト) | `SQLInsertRowsOperator` | 行 (または XCom) を `rows` として渡し、`columns` でカラム名を指定し、`insert_args={"executemany": True}` を設定して `clickhouse-connect` のネイティブ insert を使用します。あるいは `@task` から `ClickHouseHook.bulk_insert_rows` を呼び出します。[ステップ5](#step-5-clickhousehookexecute-to-dbapihook-methods)を参照してください。 | | `settings` | `hook_params={"session_settings": {...}}` | `hook_params` を通じてテンプレート化されます。値は connection の `session_settings` に上書きする形でマージされます。 | | `do_xcom_push` | `do_xcom_push` | フラグは同じですが、複数ステートメントの task では shape が異なります。plugin は最後のステートメントの結果のみを push していました。ステートメントのリストを指定した場合、provider はステートメントごとに1エントリを持つリストを push します。[複数ステートメントの結果](#multi-statement-results)を参照してください。 | | `query_id` | `hook_params={"session_settings": {"query_id": "..."}}` | `hook_params` を通じてテンプレート化されます。plugin と同様、複数ステートメントの task では、すべてのステートメントに同じ ID が送信されます。未設定の場合、`clickhouse-connect` はステートメントごとに一意の ID を生成します。`ハンドラー` を使えば `cursor.summary[-1]["query_id"]` から最後の ID を読み取れます。 | | `with_column_types` | `ハンドラー` | 行と併せて `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}}, ) ```

複数ステートメントの結果

このpluginは**最後**のステートメントの結果をXComにプッシュしていました。`SQLExecuteQueryOperator`は`sql`がリストの場合、ステートメントごとに1つの結果を返します。そのため、上記の例では`[[], [(12345.0,)]]`がプッシュされますが、pluginでは`[(12345.0,)]`がプッシュされていました。次のいずれかの対応を行ってください: * ダウンストリームの`xcom_pull`を変更し、最後のelementを取得するようにする。 * ステートメントを`;`で区切った単一の文字列として渡し、`split_statements=True`を設定する。デフォルトの`return_last=True`のままであれば、operatorは最後のステートメントの行のみをプッシュするため、pluginと同じ挙動になります。 ```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, ) ``` `parameters` を使って行のリストを挿入していた `ClickHouseOperator` は、 `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` を渡してください。指定しない場合、operator は SQLAlchemy 経由でテーブルを参照します。

カラム型を保持する

`with_column_types=True` は `(rows, [(name, type), ...])` を返していました。これはハンドラーを使って再現できます。`clickhouse-connect` のカーソルは、`cursor.description` で ClickHouse の型名を返します: ```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 は、`clickhouse_driver.Client.execute` をそのまま反映した `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` およびその他のハンドラーは `airflow.providers.common.sql.hooks.handlers` から import します。 プラグイン時代のコードで最も多く見られる hook の書き方は bulk insert です。これは単純な `run` 呼び出しには置き換えられません。DB-API のカーソルが行を SQL 文字列へ整形しようとしてしまうためです。代わりに ネイティブ insert を使用してください: 変更前: ```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)` も最終的にはネイティブ insert になりますが、それは `executemany=True` の場合のみで、デフォルトでは 1 行ごとに 1 回の HTTP リクエストを送信します。 DB-API が提供する範囲でカバーできない操作については、`get_client()` が Airflow connection の設定を反映した生の `clickhouse-connect` client を返します。これは、plugin が公開していた `clickhouse-driver` 固有の argument すべてに代わるものです: ```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` へ

置き換えのなかで、callable への入力が変わるのはこれだけです。 | | `ClickHouseSensor` | `SqlSensor` | | - | - | - | | 成功判定の callable | `is_success(result)` | `success(cell)` | | 失敗判定の callable | `is_failure(result)` | `failure(cell)` | | callable が受け取るもの | 最後のステートメントの結果全体 (row tuples の `list`) | 既定では最初の行の最初のカラム (`selector=itemgetter(0)`) | | 既定の成功条件 | `bool(result)`。行が返ってきた場合に true | 行が返ってきた場合は `bool(first cell)`、返ってこなかった場合は `False` | | `sql` | 文の文字列またはリスト、および `ClickHouseOperator` のすべての引数 | 単一の文字列。database と session settings には `hook_params` を使用します。 | | 接続 id | `clickhouse_conn_id` (既定値は `clickhouse_default`) | `conn_id` (必須) | | 行が返らない場合 | `is_success([])`。既定では `False` | `False`、または `fail_on_empty=True` の場合は error | plugin は result set 全体を渡していたため、既存のセンサーのコードはそこに索引でアクセスしています。移行時には、この索引アクセスを削除してください。 変更前: ```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` プレフィックス付きラッパーを使用していたコードは、最も作業が少なくて済みます: * import を `common.sql` モジュールに変更し、クラス名から `ClickHouse` プレフィックスを削除します。 * `conn_id` を明示的に渡します。ラッパーは `conn_id` が未指定または `None` の場合を `clickhouse_default` として扱っていましたが、`common.sql` のクラスにはデフォルトがないため、指定しないと失敗します。 `default_args={"conn_id": "clickhouse_default"}` を指定すればDAG全体をカバーできます。 * オペレーターの `database=` およびセンサーの `hook_params={"schema": ...}` は引き続き動作します。 プロバイダーのフックは `schema` を `database` の別名として扱います。 * `ClickHouseDbApiHook` は `ClickHouseHook` になります。コンストラクター引数 `schema` は引き続き別名として 受け付けられます。`database` がClickHouseネイティブの表記であり、両方が指定された場合はこちらが優先されます。 * 接続には、ステップ2でのポートおよびextrasの変更が引き続き必要です。ラッパーもネイティブ プロトコルを使用していたためです。

確認すべき動作の違い

コードのコンパイルが通った後でも、実行時の動作が異なる点がいくつかあります。 **INSERT タスクの XCom 値。** プラグインは `clickhouse-driver` が返した値をそのまま push しており、パラメータ付きの `VALUES` 挿入では挿入された行数が返されていました。一方 provider は、行を返さないステートメントに対しては空の結果セットを push します。XCom から行数を読み取っている下流のタスクは、後続の `SELECT count()` を実行するなど、別の方法で行数を取得する必要があります。 **セッション設定と `SET` ステートメント。** どちらのパッケージも複数ステートメントのリストを単一の接続上で実行し、`clickhouse-connect` はデフォルトでクライアントごとにセッションを作成するため、リストの先頭付近にある `SET` ステートメントは後続のステートメントにも引き続き適用されます。それでも `session_settings` の使用を推奨します。明示的でテンプレート化も可能であり、サーバーや中間の proxy がセッションを保持するかどうかに関わらず同じように動作します。DAG が `SET` に依存している場合は、ご自身の環境で動作を確認してください。 **例外。** エラーは `clickhouse_driver.errors.ServerException` や `NetworkError` ではなく、`clickhouse_connect.driver.exceptions.DatabaseError`、`OperationalError`、`ProgrammingError` になりました。`except` 句、`on_failure_callback` のコード、例外の型を判別するリトライロジックを更新してください。 **圧縮。** `clickhouse-driver` は `compression` が設定されていない限り圧縮を無効にしていましたが、`clickhouse-connect` はデフォルトで HTTP レスポンスの圧縮を有効にし、アルゴリズムをサーバーとネゴシエートします。以前の動作に戻すには、接続の extra に `"compress": false` を設定してください。ネイティブ圧縮のためにプラグインが必要としていた `clickhouse-cityhash` パッケージはもう不要です。`lz4` は `clickhouse-connect` の依存関係として引き続きインストールされます。 **型マッピング。** どちらのドライバーもネイティブな Python の型を返しますが、これらは別々のコードベースです。タイムゾーン付きの `DateTime64`、`Decimal`、`UUID`、`Nullable` カラム、ネストした `Array` や `Map` の値について、厳密な型に依存しているタスクを確認してください。特に結果が XCom に push されて下流で利用される場合は注意が必要です。 **`system.query_log` でのクエリの識別。** クエリは HTTP インターフェイス経由で届くようになったため、`interface = 1` ではなく `interface = 2` として記録されます。また [`system.query_log`](/ja/reference/system-tables/query_log) の `http_user_agent` カラムには、Airflow と provider のバージョン、および `client_name` extra が設定されていればその値が含まれます。ネイティブプロトコルや `clickhouse-driver` のクライアント名でフィルタしていた監視は更新が必要です。行を返さない `SELECT` では 2 つ目のエントリが生成されます。これは DB-API カーソルがカラムのメタデータを取得するために `SELECT * FROM (...) LIMIT 0` を実行するためです。 **HTTP 経由のタイムアウト。** `send_receive_timeout` は HTTP の読み取りタイムアウトになり、worker と ClickHouse の間にある proxy やロードバランサーはリクエストに独自のアイドルタイムアウトを適用します。ネイティブプロトコル上で数分間実行されていたステートメントでは、これらの上限を引き上げる必要があるかもしれません。 **接続の扱い。** hook は `run` または `get_client` の呼び出しごとに `clickhouse-connect` クライアントを作成します。これは、プラグインが `execute` ごとに新しいネイティブ接続を開いていたのと同じ考え方です。クライアントはプロセス全体で HTTP 接続プールを共有するため、`get_client()` から得たクライアントで `close()` を呼ぶことは必須ではなく、あくまで良い作法にすぎません。プールはタスクのプロセスが終了した時点で解放されます。クライアントはコンテキストマネージャーなので、`with hook.get_client() as client:` と書くのが最もすっきりします。

チェックリスト

1. Airflow が 2.11 以降であること。 2. HTTP ポートに worker から到達できること。HTTP エンドポイント用の TLS 証明書が有効であること。 3. すべての ClickHouse connection について、type が `clickhouse`、ポートが `8123` または `8443`、extras がステップ 2 に従って変換済みで、`airflow connections test` による検証が済んでいること。 4. ステップ 3 に従って import を置き換え、`common.sql` の wrapper から `ClickHouse` プレフィックスを削除済みであること。 5. operator と sensor の `clickhouse_conn_id` を `conn_id` にリネームし、plugin のデフォルトに依存していたすべての task に `conn_id` を設定済みであること。 6. `settings=` を `hook_params={"session_settings": ...}` に移動済みであること。 7. `hook.execute("INSERT ... VALUES", rows)` を `bulk_insert_rows` に、`ClickHouseOperator(parameters=rows)` を `SQLInsertRowsOperator` に置き換え済みであること。 8. sensor の callable を `result[0][0]` ではなく cell の値そのものを参照するよう調整済みであること。 9. `with_column_types`、`external_tables`、`columnar`、`query_id`、`types_check` の使用箇所を `ハンドラー` または `get_client()` を使って書き換え済みであること。 10. 複数 statement の task および INSERT task が返す XCom を利用する後続のコンシューマーを見直し済みであること。 11. `clickhouse_driver` の exception をキャッチしているコードを更新済みであること。 12. `airflow-clickhouse-plugin` と `clickhouse-driver` をアンインストール済みであること。

AIコーディングアシスタントを使う

上記の Mapping は意図的に機械的な内容にしてあるため、コーディングアシスタントに 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. ``` 差分を確認してください。connection の変更、センサーのセマンティクス、XCom のコンシューマーは、自動書き換えが失敗しやすい箇所です。

関連コンテンツ

* [Apache Airflow を ClickHouse に接続する](/ja/integrations/connectors/data-ingestion/etl-tools/airflow-and-clickhouse) * [`clickhouse-connect` Python クライアント](/ja/integrations/language-clients/python/index) * [`apache-airflow-providers-clickhousedb` リファレンスドキュメント](https://airflow.apache.org/docs/apache-airflow-providers-clickhousedb/) * [GitHub 上の airflow-clickhouse-plugin](https://github.com/bryzgaloff/airflow-clickhouse-plugin)