高度な使い方

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

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

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

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

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

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 接続プールです。

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 で実行される場合があります。

非同期 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 クエリは、ClickHouse の「session」のコンテキスト内で実行されます。現在、session は次の 2 つの目的で使用されます。

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

デフォルトでは、同期 Client は生成された session ID を使用します。そのため、SET ステートメントと一時テーブルは、そのクライアントからのリクエスト間で保持されます。async ファクトリーは、デフォルトでは session ID を生成しません。ClickHouse では同じ session 内でクエリを同時実行できず、これを試みるとクライアントは ProgrammingError を送出するため、次のいずれかのパターンを使用してください。

1. session の分離が必要な各スレッド / プロセス / イベントハンドラーごとに、個別の Client インスタンスを作成します。これにより、クライアントごとの session 状態 (一時テーブルと SET 値) が維持されます。
2. 共有session状態が不要な場合は、query、command、または insert の呼び出し時に settings 引数を使用して、各クエリに一意の session_id を指定します。
3. 共有クライアントで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 レベルの設定は、リクエストをまたいで保持されません。

ClickHouse Connect は、サーバーとの基盤となる HTTP 接続を処理するために urllib3 の接続プールを使用します。デフォルトでは、すべてのクライアントインスタンスが同じ接続プールを共有しており、ほとんどのユースケースではこれで十分です。このデフォルトのプールでは、アプリケーションで使用される各 ClickHouse サーバーに対して、最大 8 本の HTTP Keep-Alive 接続が維持されます。

大規模なマルチスレッドアプリケーションでは、接続プールを分けたほうが適切な場合があります。カスタマイズした接続プールは、メインの 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を参照してください。

async クライアントは urllib3 を使用せず、aiohttp のプールを使用します。設定は、get_async_client の connector_limit、connector_limit_per_host、keepalive_timeout で行います。await async_client.close_connections() を呼び出すと、進行中のリクエストを中断することなくプールが入れ替わります。