高度な使い方

Raw API

ClickHouse のデータとネイティブまたはサードパーティのデータ型・構造との間で変換が不要なユースケースでは、ClickHouse Connect クライアントは ClickHouse 接続を直接利用するためのメソッドを提供します。

Client `raw_query` メソッド

Client.raw_query メソッドを使用すると、クライアント接続を通じて ClickHouse の HTTP クエリインターフェイスを直接利用できます。戻り値は未処理の bytes オブジェクトです。このメソッドは、パラメータバインディング、エラーハンドリング、再試行、設定管理を最小限のインターフェイスで扱える便利なラッパーを提供します。

Parameter	Type	Default	Description
`query`	str	Required	任意の有効な ClickHouse クエリ。
`parameters`	dict or sequence	`None`	Parameters argument を参照してください。
`settings`	dict	`None`	Settings argument を参照してください。
`fmt`	str	`None`	結果として返される bytes に使用する ClickHouse の出力フォーマットです。 (指定しない場合、ClickHouse は TSV を使用します)
`use_database`	bool	`True`	クエリコンテキストで、クライアントに設定されたデータベースを使用します。
`external_data`	`ExternalData`	`None`	クエリで使用する外部ファイルまたはバイナリデータです。External data を参照してください。
`transport_settings`	dict	`None`	このリクエストに追加される HTTP headers です。

結果として返される bytes オブジェクトの処理は呼び出し元の責任です。なお、Client.query_arrow は、このメソッドを ClickHouse の Arrow 出力フォーマットで利用するごく薄いラッパーにすぎません。

Client `raw_stream` メソッド

同期版の Client.raw_stream メソッドは raw_query と同じ API ですが、バイト chunk の io.IOBase ストリームを返します。処理が完了したら、ストリームを閉じてください。AsyncClient.raw_stream は await して使用し、async with と async for で使用する async StreamContext を返します。

Client `raw_insert` メソッド

Client.raw_insert メソッドを使うと、クライアント接続を介して bytes オブジェクトまたは bytes オブジェクトを生成するジェネレーターを直接 insert できます。insert payload の処理を行わないため、非常に高いパフォーマンスを発揮します。このメソッドには、settings と insert format を指定するオプションがあります。

Parameter	Type	Default	Description
`table`	str	Required	単純な table 名、または database 修飾付きの table 名。
`column_names`	Sequence[str]	`None`	insert block のカラム名。`fmt` に名前が含まれていない場合は必須です。
`insert_block`	str, bytes, generator, or `BinaryIO`	Required	insert するデータ。String はクライアントの encoding を使用してエンコードされます。
`settings`	dict	`None`	Settings argument を参照してください。
`fmt`	str	`None`	`insert_block` payload の ClickHouse input format です。フォーマットを指定しない場合は `Native` が使用されます。
`compression`	str	`None`	`"gzip"`、`"lz4"`、`"zstd"` など、`insert_block` にすでに適用されている圧縮。
`transport_settings`	dict	`None`	このリクエストに追加される HTTP headers。

insert_block が指定されたフォーマットで、指定された圧縮 method を使用していることを保証する責任は呼び出し元にあります。ClickHouse Connect は、ファイルのアップロードや PyArrow Tables に対してこれらの raw insert を使用し、パースは ClickHouse server に委ねます。

クエリ結果をファイルとして保存する

raw_stream メソッドを使うと、ClickHouse からローカルファイルシステムへファイルを直接ストリーミングできます。たとえば、クエリ結果を CSV ファイルとして保存するには、次のコードスニペットを使用します。

import clickhouse_connect

if __name__ == "__main__":
    client = clickhouse_connect.get_client()
    query = (
        "SELECT number, toString(number) AS number_as_str "
        "FROM system.numbers LIMIT 5"
    )
    stream = client.raw_stream(query=query, fmt="CSVWithNames")
    try:
        with open("output.csv", "wb") as file:
            for chunk in stream:
                file.write(chunk)
    finally:
        stream.close()
        client.close()

上記のコードを実行すると、次の内容を含むoutput.csvファイルが生成されます。

"number","number_as_str"
0,"0"
1,"1"
2,"2"
3,"3"
4,"4"

同様に、データは TabSeparated やその他のフォーマットで保存することもできます。利用可能なすべてのフォーマットオプションの概要については、Formats for Input and Output Data を参照してください。

マルチスレッド、マルチプロセス、非同期/イベント駆動のユースケース

ClickHouse Connect は、マルチスレッド、マルチプロセス、イベントループ駆動/非同期のアプリケーションでも問題なく動作します。すべてのクエリ処理と insert 処理は単一のスレッド内で行われるため、操作は通常スレッドセーフです。 (単一スレッドによる性能面の不利を補うため、将来的には一部の操作で低レベルの並列処理が導入される可能性がありますが、その場合でもスレッドセーフ性は維持されます。) 実行される各クエリまたは insert は、それぞれ専用の QueryContext または InsertContext オブジェクトに状態を保持するため、これらのヘルパーオブジェクトはスレッドセーフではなく、複数の処理ストリーム間で共有すべきではありません。コンテキストオブジェクトの詳細については、QueryContexts および InsertContexts の各セクションを参照してください。さらに、同時に 2 つ以上のクエリや insert が「進行中」のアプリケーションでは、もう 2 つ注意すべき点があります。1 つ目はクエリ/insert に関連付けられる ClickHouse の「session」で、2 つ目は ClickHouse Connect Client のインスタンスで使用される HTTP 接続プールです。

AsyncClient

ClickHouse Connect は、asyncio アプリケーション向けに aiohttp ベースのネイティブクライアントを提供しています。使用する前に、オプションの依存関係をインストールしてください。

pip install "clickhouse-connect[async]"

get_async_client を await して、クライアントを作成・初期化します。query、command、insert などの I/Oメソッドはコルーチンです。

import asyncio

import clickhouse_connect


async def main():
    async with await clickhouse_connect.get_async_client() as client:
        result = await client.query(
            "SELECT name FROM system.databases ORDER BY name LIMIT 1"
        )
        print(result.result_rows)


asyncio.run(main())

非同期クライアントは、同期クライアントと同じ query、insert、raw、Arrow、streaming のインターフェースに従います。ネットワーク I/O には aiohttp を使用します。CPU 負荷の高い Native フォーマットのパース処理は、イベントループをブロックしないように executor で実行される場合があります。非同期クライアントは、特定のイベントループ内で作成された aiohttp セッションを保持します。クライアントを別のイベントループに移す場合は、まず所有元のループ内でクライアントをクローズし、新しいループでリクエストを行う前に await client._initialize() を呼び出してください。所有元のループがすでにクローズされている場合は、現在のループ内で await client.close() を呼び出してから await client._initialize() を呼び出します。所有元のループがクローズされた後にクリーンアップを開始すると、aiohttp がクローズされていないトランスポートについて警告を出すことがあるため、可能な限り移行前にクローズしてください。非同期 streaming メソッドは、返されたコンテキストに入る前に await されます:

async with await client.query_rows_stream(
    "SELECT number FROM numbers(100000)"
) as stream:
    async for row in stream:
        process(row)

同期ファクトリーとは異なり、get_async_client では、同時実行するコルーチン間でクライアントを共有できるよう、デフォルトで session ID の自動生成が無効になっています。明示的な session_id または autogenerate_session_id=True を渡すのは、session 状態が必要で、かつその session 内で同時実行クエリを行わない場合に限ってください。

ClickHouse session ID の管理

各 ClickHouse クエリは、ClickHouse の「session」のコンテキスト内で実行されます。現在、session は次の 2 つの目的で使用されます。

複数のクエリに特定の ClickHouse settings を関連付けるため (ユーザー設定を参照) 。ClickHouse の SET コマンドは、ユーザーsessionの範囲で設定を変更するために使用されます。
一時テーブルを追跡するため

デフォルトでは、同期 Client は生成された session ID を使用します。SET ステートメントと一時テーブルがそのクライアントからのリクエスト間で保持されるのは、それらのリクエストが同じ ClickHouseサーバープロセスに到達した場合に限られます。async ファクトリーは、デフォルトでは session ID を生成しません。名前付き session の状態と同一 session の重複チェックはプロセスローカルであり、クライアントはリクエスト送信前にローカルで重複を検出すると ProgrammingError を送出します。ClickHouse Cloud やその他のロードバランサーを介したデプロイ環境では、固定の session_id を分散状態や分散ミューテックスとして利用しないでください。重複が問題となる場合は、ClickHouse にリクエストを送信する前に直列化してください。次のいずれかのパターンを使用してください。

session の分離が必要な各スレッド / プロセス / イベントハンドラーごとに、個別の Client インスタンスを作成します。これにより、クライアントごとの session 状態 (一時テーブルと SET 値) が維持されます。
共有session状態が不要な場合は、query、command、または insert の呼び出し時に settings 引数を使用して、各クエリに一意の session_id を指定します。
共有クライアントでsessionを無効にするには、クライアントを作成する前に autogenerate_session_id=False を設定します (または、これを直接 get_client に渡します) 。

import clickhouse_connect
from clickhouse_connect import common

common.set_setting("autogenerate_session_id", False)
client = clickhouse_connect.get_client(
    host="somehost.com",
    username="dbuser",
    password="password",
)

あるいは、autogenerate_session_id=False を get_client(...) に直接渡します。この場合、ClickHouse Connect は session_id を送信しないため、server は個々のリクエストを同じ session に属するものとして扱いません。一時テーブルや session レベルの設定は、リクエストをまたいで保持されません。

HTTP接続プールのカスタマイズ

ClickHouse Connect は、サーバーとの基盤となる HTTP 接続を処理するために urllib3 の接続プールを使用します。デフォルトでは、プロセス内のすべての同期クライアントインスタンスが同じ接続プールを共有しており、ほとんどのユースケースではこれで十分です。multiprocessing の各ワーカーは、それぞれプロセスローカルなデフォルトのプールを持ち、そのワーカー内で作成されたクライアント間で再利用します。フォーク前に作成されたクライアントは親プロセスのプールを保持したままとなるため、子プロセスでは使用しないでください。デフォルトのプールでは、アプリケーションで使用される各 ClickHouseサーバーに対して、最大 8 本の HTTP Keep-Alive 接続が維持されます。デフォルトのソケットオプションでは、TCP キープアライブと TCP_NODELAY が有効になっています。ソケットの送信・受信バッファサイズは、オペレーティングシステムによって管理されます。大規模なマルチスレッドアプリケーションでは、接続プールを分けたほうが適切な場合があります。カスタマイズした接続プールは、メインの clickhouse_connect.get_client 関数に pool_mgr キーワード引数として指定できます：

import clickhouse_connect
from clickhouse_connect.driver import httputil

big_pool_mgr = httputil.get_pool_manager(maxsize=16, num_pools=12)

client1 = clickhouse_connect.get_client(pool_mgr=big_pool_mgr)
client2 = clickhouse_connect.get_client(pool_mgr=big_pool_mgr)

クライアントでプールマネージャーを共有することも、各クライアントが個別のマネージャーを使用することもできます。詳細については、urllib3 PoolManager documentationを参照してください。ソケットオプションを設定するには、httputil.get_pool_manager または httputil.get_pool_manager_options に socket_options を渡します。この場合、キープアライブのオプションや TCP_NODELAY を含むデフォルトのリスト全体が置き換えられます。ソケットオプションを明示的に一切指定しない場合は、[] または None を渡してください。非同期クライアントは urllib3 を使用せず、aiohttp のプールを使用します。設定は、get_async_client の connector_limit、connector_limit_per_host、keepalive_timeout で行います。await async_client.close_connections() を呼び出すと、進行中のリクエストを中断することなくプールが入れ替わります。 async のクエリおよび挿入では、プールの空きスロットの待機にタイムアウトはありません。ストリーミングレスポンスは、最後まで読み取るかクローズして、プールのスロットを解放してください。connect_timeout はスロットが利用可能になった時点から計測が始まり、DNS 名前解決、TCP および TLS のセットアップ、プロキシとのネゴシエーションが対象となります。send_receive_timeout はソケットからの読み取りに適用されます。プールでの待機を含む操作全体にデッドラインを設定するには、asyncio.wait_for を使用します。例：await asyncio.wait_for(client.query("SELECT 13"), timeout=30)。

ホーム

JavaScript

Python

Go

Java

C++

C

サードパーティ製クライアント

Raw API

Client `raw_query` メソッド

Client `raw_stream` メソッド

Client `raw_insert` メソッド

クエリ結果をファイルとして保存する

マルチスレッド、マルチプロセス、非同期/イベント駆動のユースケース

AsyncClient

ClickHouse session ID の管理

HTTP接続プールのカスタマイズ

​Raw API

​Client raw_query メソッド

​Client raw_stream メソッド

​Client raw_insert メソッド

​クエリ結果をファイルとして保存する

​マルチスレッド、マルチプロセス、非同期/イベント駆動のユースケース

​AsyncClient

​ClickHouse session ID の管理

​HTTP接続プールのカスタマイズ

Raw API

Client `raw_query` メソッド

Client `raw_stream` メソッド

Client `raw_insert` メソッド

クエリ結果をファイルとして保存する

マルチスレッド、マルチプロセス、非同期/イベント駆動のユースケース

AsyncClient

ClickHouse session ID の管理

HTTP接続プールのカスタマイズ