高级用法

底层 API

对于无需在 ClickHouse 数据与原生或第三方数据类型及结构之间进行转换的场景，ClickHouse Connect 客户端提供了一些方法，可直接使用 ClickHouse 连接。

客户端 `raw_query` 方法

Client.raw_query 方法允许通过客户端连接直接使用 ClickHouse HTTP 查询接口。返回值是一个未经处理的 bytes 对象。它通过极简接口提供了便捷封装，支持参数绑定、错误处理、重试和 settings 管理：

参数	类型	默认值	说明
`query`	str	必填	任何有效的 ClickHouse 查询。
`parameters`	dict or sequence	`None`	请参见Parameters 参数。
`settings`	dict	`None`	请参见Settings 参数。
`fmt`	str	`None`	ClickHouse 输出格式。未指定格式时，ClickHouse 使用 TSV。
`use_database`	bool	`True`	包含客户端上配置的数据库。
`external_data`	`ExternalData`	`None`	外部文件或二进制数据。请参见外部数据。
`transport_settings`	dict	`None`	添加到此请求的 HTTP 请求头。

如何处理返回的 bytes 对象由调用方负责。请注意，Client.query_arrow 只是对此方法的一个轻量封装，使用的是 ClickHouse Arrow 输出格式。

Client `raw_stream` 方法

同步 Client.raw_stream 方法与 raw_query 的 API 相同，但会返回一个由字节块组成的 io.IOBase 流。处理完成后请关闭该流。AsyncClient.raw_stream 需要使用 await 等待，并会返回一个可与 async with 和 async for 搭配使用的异步 StreamContext。

Client `raw_insert` 方法

Client.raw_insert 方法允许通过客户端连接直接插入 bytes 对象或 bytes 对象生成器。由于它不会对插入载荷进行任何处理，因此性能非常高。该方法提供了用于指定设置和插入格式的选项：

Parameter	Type	Default	Description
`table`	str	必填	简单表名或带数据库限定名的目标表。
`column_names`	Sequence[str]	`None`	插入块的列名。当 `fmt` 不包含列名时为必填项。
`insert_block`	str, bytes, generator, or `BinaryIO`	必填	要插入的数据。字符串会使用客户端编码进行编码。
`settings`	dict	`None`	请参见 Settings 参数。
`fmt`	str	`None`	`insert_block` 载荷的 ClickHouse 输入格式。未指定格式时使用 `Native`。
`compression`	str	`None`	已应用于 `insert_block` 的压缩，例如 `"gzip"`、`"lz4"` 或 `"zstd"`。
`transport_settings`	dict	`None`	添加到此请求的 HTTP 请求头。

调用方有责任确保 insert_block 采用指定的格式，并使用指定的压缩方法。ClickHouse Connect 会将这些原始插入用于文件上传和 PyArrow 表，并将解析工作交由 ClickHouse 服务器处理。

将查询结果保存为文件

你可以使用 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 等其他格式。有关所有可用格式选项的概述，请参阅输入和输出数据格式。

多线程、多进程和异步/事件驱动用例

ClickHouse Connect 非常适合用于多线程、多进程以及事件循环驱动/异步应用程序。所有查询和插入处理都在单个线程内完成，因此这些操作通常是线程安全的。 (未来可能会在较低层级为某些操作引入并行处理，以克服单线程带来的性能损耗；但即便如此，线程安全性仍会得到保证。) 由于每个已执行的查询或插入都会分别在各自的 QueryContext 或 InsertContext 对象中维护状态，这些辅助对象本身并不是线程安全的，因此不应在多个处理流之间共享。有关上下文对象的更多讨论，请参见 QueryContexts 和 InsertContexts 部分。此外，对于同时有两个或更多查询和/或插入“在进行中”的应用程序，还需要注意另外两个方面。第一是与查询/插入关联的 ClickHouse“会话”，第二是 ClickHouse Connect Client 实例使用的 HTTP 连接池。

AsyncClient

ClickHouse Connect 为 asyncio 应用程序提供了一个基于 aiohttp 的原生客户端。使用前，请先安装可选依赖项：

pip install "clickhouse-connect[async]"

等待 get_async_client 完成以创建并初始化客户端。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())

异步客户端与同步客户端遵循相同的查询、插入、raw、Arrow 和流式约定。它使用 aiohttp 进行网络 I/O。受 CPU 限制的 Native 格式解析可能会在 executor 中运行，以免阻塞事件循环。异步客户端持有一个 aiohttp 会话，该会话创建于某个特定的事件循环中。若要将客户端移至另一个事件循环，请先在其所属的循环中将其关闭，然后在新循环中调用 await client._initialize()，再发起请求。如果所属的循环已经关闭，请在当前循环中依次调用 await client.close() 和 await client._initialize()。如果清理操作在所属循环关闭之后才开始，aiohttp 仍可能报告存在未关闭的 transport，因此请尽量在迁移之前先关闭客户端。在进入返回的上下文之前，需要先 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 默认禁用自动生成会话 ID，以便并发协程共享同一个客户端。只有在你需要会话状态，且能够避免在该会话中并发执行查询时，才应显式传入 session_id 或设置 autogenerate_session_id=True。

管理 ClickHouse 会话 ID

每个 ClickHouse 查询都在 ClickHouse “会话”的上下文中执行。目前，会话主要有两个用途：

将特定的 ClickHouse settings 关联到多个查询 (请参见用户 settings) 。ClickHouse SET 命令用于在用户会话范围内更改 settings。
跟踪临时表

默认情况下，同步 Client 会使用自动生成的会话 ID。只有当来自该客户端的请求到达同一个 ClickHouse 服务器进程时，SET 语句和临时表才会在这些请求之间保留。async factory 默认不会生成会话 ID。命名会话的状态以及同一会话的重叠检查仅在本进程内有效；如果客户端在发送请求之前检测到本地重叠，会引发 ProgrammingError。在 ClickHouse Cloud 或其他采用负载均衡的部署中，不要依赖固定的 session_id 作为分布式状态或分布式互斥锁。如果重叠会带来问题，请在向 ClickHouse 发送请求之前自行对请求进行序列化。请使用以下模式之一：

为每个需要会话隔离的线程/进程/event handler 创建单独的 Client 实例。这样可以保留每个客户端各自的会话状态 (临时表和 SET 值) 。
如果不需要共享会话状态，请在调用 query、command 或 insert 时，通过 settings argument 为每个查询指定唯一的 session_id。
通过在创建客户端之前设置 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；服务器不会将不同的请求视为属于同一会话。临时表和会话级设置不会在请求之间保留。

自定义 HTTP 连接池

ClickHouse Connect 使用 urllib3 连接池来管理与服务器之间的底层 HTTP 连接。默认情况下，同一进程中的所有同步客户端实例共享同一个连接池，这对于绝大多数使用场景已经足够。每个 multiprocessing 工作进程都有各自的进程本地默认连接池，并在该进程内创建的所有客户端之间复用。在 fork 之前创建的客户端会保留父进程的连接池，不应在子进程中使用。该默认连接池会为应用程序使用的每个 ClickHouse 服务器维护最多 8 个 HTTP Keep-Alive 连接。默认的套接字选项会启用 TCP keepalive 和 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 文档。要设置套接字选项，请将 socket_options 传给 httputil.get_pool_manager 或 httputil.get_pool_manager_options。这会替换整个默认选项列表，包括 keepalive 选项和 TCP_NODELAY。传入 [] 或 None 则表示不设置任何显式套接字选项。 async 客户端使用的是 aiohttp 连接池，而不是 urllib3。可通过 get_async_client 上的 connector_limit、connector_limit_per_host 和 keepalive_timeout 对其进行配置。调用 await async_client.close_connections() 会轮换连接池，而不会中断正在处理中的请求。对于异步查询和插入，等待空闲连接池槽位时不设超时。请完整读取或关闭流式响应，以释放其占用的连接池槽位。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

第三方客户端

底层 API

客户端 `raw_query` 方法

Client `raw_stream` 方法

Client `raw_insert` 方法

将查询结果保存为文件

多线程、多进程和异步/事件驱动用例

AsyncClient

管理 ClickHouse 会话 ID

自定义 HTTP 连接池

​底层 API

​客户端 raw_query 方法

​Client raw_stream 方法

​Client raw_insert 方法

​将查询结果保存为文件

​多线程、多进程和异步/事件驱动用例

​AsyncClient

​管理 ClickHouse 会话 ID

​自定义 HTTP 连接池

底层 API

客户端 `raw_query` 方法

Client `raw_stream` 方法

Client `raw_insert` 方法

将查询结果保存为文件

多线程、多进程和异步/事件驱动用例

AsyncClient

管理 ClickHouse 会话 ID

自定义 HTTP 连接池