> ## Documentation Index
> Fetch the complete documentation index at: https://clickhouse.com/docs/llms.txt
> Use this file to discover all available pages before exploring further.

> ClickHouse provider を使用して、Apache Airflow から ClickHouse のクエリ実行とデータロードをオーケストレーションします

# Apache Airflow を ClickHouse に接続

export const ClickHouseSupportedBadge = () => {
  return <div className="ClickHouseSupportedBadge">
            <div className="ClickHouseSupportedIcon">
                <svg width="16" height="16" viewBox="0 0 16 16" fill="none" xmlns="http://www.w3.org/2000/svg">
                    <path d="M1.30762 1.39073C1.30762 1.3103 1.37465 1.22986 1.46849 1.22986H2.64824C2.72868 1.22986 2.80912 1.29689 2.80912 1.39073V14.4886C2.80912 14.5691 2.74209 14.6495 2.64824 14.6495H1.46849C1.38805 14.6495 1.30762 14.5825 1.30762 14.4886V1.39073Z" fill="currentColor" />
                    <path d="M4.2832 1.39073C4.2832 1.3103 4.35023 1.22986 4.44408 1.22986H5.62383C5.70427 1.22986 5.7847 1.29689 5.7847 1.39073V14.4886C5.7847 14.5691 5.71767 14.6495 5.62383 14.6495H4.44408C4.36364 14.6495 4.2832 14.5825 4.2832 14.4886V1.39073Z" fill="currentColor" />
                    <path d="M7.25977 1.39073C7.25977 1.3103 7.3268 1.22986 7.42064 1.22986H8.60039C8.68083 1.22986 8.76127 1.29689 8.76127 1.39073V14.4886C8.76127 14.5691 8.69423 14.6495 8.60039 14.6495H7.42064C7.3402 14.6495 7.25977 14.5825 7.25977 14.4886V1.39073Z" fill="currentColor" />
                    <path d="M10.2354 1.39073C10.2354 1.3103 10.3024 1.22986 10.3962 1.22986H11.576C11.6564 1.22986 11.7369 1.29689 11.7369 1.39073V14.4886C11.7369 14.5691 11.6698 14.6495 11.576 14.6495H10.3962C10.3158 14.6495 10.2354 14.5825 10.2354 14.4886V1.39073Z" fill="currentColor" />
                    <path d="M13.2256 6.6057C13.2256 6.52526 13.2926 6.44482 13.3865 6.44482H14.5662C14.6466 6.44482 14.7271 6.51186 14.7271 6.6057V9.27354C14.7271 9.35398 14.6601 9.43442 14.5662 9.43442H13.3865C13.306 9.43442 13.2256 9.36739 13.2256 9.27354V6.6057Z" fill="currentColor" />
                </svg>
            </div>
            ClickHouse対応
        </div>;
};

<ClickHouseSupportedBadge />

[Apache Airflow](https://airflow.apache.org/) は、ワークフローをコードとして記述し、スケジュールし、監視するためのオープンソースプラットフォームです。ワークフローは、Python で記述されたタスクの有向非巡回グラフ (DAG) として定義されます。

`apache-airflow-providers-clickhousedb` プロバイダーは Airflow を ClickHouse に接続し、DAG の一部としてクエリの実行、テーブルの作成、データの読み込みを行えるようにします。[HTTP インターフェイス](/docs/ja/concepts/features/interfaces/http) 経由で [`clickhouse-connect`](/docs/ja/integrations/language-clients/python/index) クライアントを使用して接続し、Airflow の共通 SQL フレームワークを通じて ClickHouse を利用できるようにするため、標準の `SQLExecuteQueryOperator` で DDL、DML、分析クエリを処理でき、ClickHouse 専用のオペレーターは不要です。

<div id="install-the-provider">
  ## プロバイダーをインストールする
</div>

Airflow のスケジューラとワーカーが実行される環境に、プロバイダーをインストールします。

```bash theme={null}
pip install apache-airflow-providers-clickhousedb
```

このプロバイダーは `apache-airflow-providers-common-sql` と `clickhouse-connect` に依存しており、これらもあわせてインストールされます。クエリ結果を pandas または polars の DataFrame に渡すには、オプションの extras をインストールしてください:

```bash theme={null}
pip install 'apache-airflow-providers-common-sql[pandas,polars]'
```

<div id="create-a-clickhouse-connection">
  ## ClickHouse 接続を作成する
</div>

このプロバイダーは、`clickhouse` という接続タイプを登録します。Airflow UI の **Admin > Connections** から接続を作成するか、CLI または環境変数で定義できます。

UI では、接続タイプとして **ClickHouse** を選択し、各フィールドに入力します。

| フィールド        | 説明                                                                                                | デフォルト                     |
| ------------ | ------------------------------------------------------------------------------------------------- | ------------------------- |
| **Host**     | ClickHouse server の hostname (例: `abc123.clickhouse.cloud`)                                       | `localhost`               |
| **Port**     | HTTP(S) のポート                                                                                      | `8123` (平文), `8443` (TLS) |
| **Login**    | ClickHouse の username                                                                             | `default`                 |
| **Password** | ClickHouse ユーザーの password                                                                         | (空)                       |
| **Database** | 接続の default database。UI では **Database** と表示されます。URI または JSON で接続を定義する場合、これは `schema` フィールドに該当します。 | `default`                 |

[ClickHouse Cloud](/docs/ja/products/cloud/getting-started/intro) または TLS が有効なセルフホスト クラスターでは、**Extra** フィールドで `secure` を `true` に設定し、TLS ポート (`8443`) を使用します。

<div id="extra-connection-options">
  ### 追加の接続オプション
</div>

このプロバイダーでは、接続フォーム内に専用フィールドとして追加オプションが用意されています。代わりに URI、JSON、または環境変数で接続を定義する場合は、これらを `extra` JSON オブジェクト内のキーとして指定してください。いずれも任意です。

| `extra` キー             | UI フィールド             | デフォルト   | 説明                                                                                                                                            |
| ---------------------- | -------------------- | ------- | --------------------------------------------------------------------------------------------------------------------------------------------- |
| `secure`               | TLS (HTTPS) を使用      | `false` | HTTPS/TLS を有効にします。                                                                                                                            |
| `verify`               | SSL 証明書を検証           | `true`  | `secure` が `true` の場合、サーバーの TLS 証明書を検証します。自己署名証明書を使用する場合は `false` に設定します。                                                                     |
| `connect_timeout`      | 接続タイムアウト (秒)         | `10`    | HTTP connection のタイムアウト時間 (秒) です。                                                                                                             |
| `send_receive_timeout` | クエリタイムアウト (秒)        | `300`   | クエリの読み取り/書き込みのタイムアウト時間 (秒) です。長時間実行される分析クエリでは、この値を増やしてください。                                                                                   |
| `compress`             | LZ4 圧縮を有効化           | `true`  | LZ4 による結果の圧縮を有効にします。                                                                                                                          |
| `client_name`          | Client Name          | (empty) | ClickHouse の `User-Agent` および [`system.query_log`](/docs/ja/reference/system-tables/query_log) の `client_name` カラムで、Airflow のバージョン識別子に付加されるラベルです。  |
| `session_settings`     | セッション設定 (JSON)       | (empty) | 接続上のすべてのクエリに適用される [ClickHouse セッション設定](/docs/ja/reference/settings/session-settings) です。たとえば `{"max_execution_time": 300, "max_threads": 8}` などです。 |
| `client_kwargs`        | Client kwargs (JSON) | (empty) | `clickhouse_connect.get_client()` に渡される追加のキーワード引数です。たとえば `http_proxy` などがあります。                                                                |

<div id="define-a-connection-without-the-ui">
  ### UI を使わずに接続を定義する
</div>

環境変数で接続を設定します。URI 形式には、ホスト、認証情報、データベースが含まれます。

```bash theme={null}
export AIRFLOW_CONN_CLICKHOUSE_DEFAULT='clickhouse://default:password@localhost:8123/my_database'
```

URI のすべての部分は URL エンコードする必要があります。TLS、タイムアウト、またはセッション設定については、**Extra** フィールドを利用できる JSON 形式を使用します。

```bash theme={null}
export AIRFLOW_CONN_CLICKHOUSE_DEFAULT='{
    "conn_type": "clickhouse",
    "host": "abc123.clickhouse.cloud",
    "port": 8443,
    "login": "default",
    "password": "secret",
    "schema": "my_database",
    "extra": {
        "secure": true,
        "session_settings": {
            "max_execution_time": 300,
            "max_memory_usage": 10000000000
        }
    }
}'
```

すべてのフックとオペレーターは、別の接続 ID を指定しない限り、接続 ID `clickhouse_default` を使用します。

<div id="run-queries">
  ## SQLExecuteQueryOperatorでクエリを実行する
</div>

オペレーターの`conn_id`をClickHouse接続に設定します。次のDAGは、テーブルを作成し、行を挿入し、それらを読み出してから、テーブルを削除します。

```python theme={null}
from datetime import datetime

from airflow import DAG
from airflow.providers.common.sql.hooks.sql import fetch_all_handler
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator

CLICKHOUSE_CONN_ID = "clickhouse_default"
CLICKHOUSE_TABLE = "airflow_example"

with DAG(
    dag_id="example_clickhouse",
    start_date=datetime(2021, 1, 1),
    default_args={"conn_id": CLICKHOUSE_CONN_ID},
    schedule="@once",
    catchup=False,
) as dag:
    create_table = SQLExecuteQueryOperator(
        task_id="create_table",
        sql=f"""
            CREATE TABLE IF NOT EXISTS {CLICKHOUSE_TABLE} (
                id   UInt32,
                name String,
                ts   DateTime DEFAULT now()
            ) ENGINE = MergeTree()
            ORDER BY id
        """,
    )

    insert_rows = SQLExecuteQueryOperator(
        task_id="insert_rows",
        sql=f"""
            INSERT INTO {CLICKHOUSE_TABLE} (id, name) VALUES
                (1, 'Alice'),
                (2, 'Bob'),
                (3, 'Charlie')
        """,
    )

    read_rows = SQLExecuteQueryOperator(
        task_id="read_rows",
        sql=f"SELECT id, name FROM {CLICKHOUSE_TABLE} ORDER BY id",
        handler=fetch_all_handler,
    )

    drop_table = SQLExecuteQueryOperator(
        task_id="drop_table",
        sql=f"DROP TABLE IF EXISTS {CLICKHOUSE_TABLE}",
    )

    create_table >> insert_rows >> read_rows >> drop_table
```

クエリ結果は、デフォルトの`handler` (`fetch_all_handler`) を使って取得されます。結果セット全体以外を返したい場合は、別のハンドラーを渡します。たとえば、最初の1行だけを返すには`fetch_one_handler`を使用します。

<div id="target-a-different-database">
  ### タスクごとに異なるデータベースを対象にする
</div>

1 つの接続先がクラスターで、各タスクが異なるデータベースにクエリを実行する場合は、接続を個別に作成するのではなく、`hook_params` でデータベースを上書きします。

```python theme={null}
read_rows = SQLExecuteQueryOperator(
    task_id="read_rows",
    conn_id=CLICKHOUSE_CONN_ID,
    sql="SELECT count() FROM events",
    hook_params={"database": "analytics"},
)
```

<div id="use-the-hook-directly">
  ## フックを直接使用する
</div>

SQLオペレーターでは対応しにくい処理 — 一括挿入、ストリーミング、または ClickHouse 固有のクライアント呼び出し — には、Python タスク内で `ClickHouseHook` を使用します。

フックの `bulk_insert_rows` メソッドは、`clickhouse-connect` のネイティブな列指向の挿入パスを使用します。これは、大規模なデータセットを1行ずつ挿入するよりもはるかに高速です。非常に大きな入力でピークメモリを抑えるには、`batch_size` を設定します：

```python theme={null}
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,
)
```

フックからは直接利用できない機能が必要な場合は、`get_client()` を呼び出して基盤となる `clickhouse-connect` クライアントにアクセスします:

```python theme={null}
client = hook.get_client()
total = client.query("SELECT count() FROM events").result_rows[0][0]
```

<div id="apply-session-settings">
  ### セッション設定を適用
</div>

フックの構築時に、[セッション設定](/docs/ja/reference/settings/session-settings) を直接、またはオペレーターの `hook_params` を通じて渡します。コンストラクターに渡した設定は、接続の **Extra** フィールドで定義された `session_settings` の上にマージされ、同じキーがある場合はコンストラクター側の値が優先されます。

```python theme={null}
hook = ClickHouseHook(
    clickhouse_conn_id="clickhouse_default",
    session_settings={"max_execution_time": 60, "max_threads": 4},
)
```

<div id="related-content">
  ## 関連コンテンツ
</div>

* [`clickhouse-connect` Python クライアント](/docs/ja/integrations/language-clients/python/index)
* [ClickHouse HTTP インターフェイス](/docs/ja/concepts/features/interfaces/http)
* [ClickHouse セッション設定リファレンス](/docs/ja/reference/settings/session-settings)
* [`apache-airflow-providers-clickhousedb` リファレンスドキュメント](https://airflow.apache.org/docs/apache-airflow-providers-clickhousedb/)
* [PyPI の Provider パッケージ](https://pypi.org/project/apache-airflow-providers-clickhousedb/)