Python 与 ClickHouse Connect 的集成

简介​

ClickHouse Connect 是一个核心数据库驱动程序，提供了与广泛 Python 应用程序的互操作性。

• 主要接口是 clickhouse_connect.driver 包中的 Client 对象。该核心包还包括各种辅助类和用于与 ClickHouse 服务器通信的实用函数，以及用于插入和选择查询的高级管理的 “context” 实现。
• clickhouse_connect.datatypes 包提供所有非实验性 ClickHouse 数据类型的基本实现和子类。其主要功能是将 ClickHouse 数据序列化和反序列化为 ClickHouse “原生” 二进制列式格式，用于实现 ClickHouse 和客户端应用程序之间的高效传输。
• clickhouse_connect.cdriver 包中的 Cython/C 类优化了一些最常见的序列化和反序列化，以显著提高纯 Python 的性能。
• 在 clickhouse_connect.cc_sqlalchemy 包中有一个有限的 SQLAlchemy 方言，它是基于 datatypes 和 dbi 包构建的。这个受限的实现专注于查询/游标功能，通常不支持 SQLAlchemy 的 DDL 和 ORM 操作。（SQLAlchemy 针对 OLTP 数据库，我们建议使用更专业的工具和框架来管理 ClickHouse OLAP 导向的数据库。）
• 核心驱动程序和 ClickHouse Connect SQLAlchemy 实现是将 ClickHouse 连接到 Apache Superset 的首选方法。使用 ClickHouse Connect 数据库连接，或 clickhousedb SQLAlchemy 方言连接字符串。

该文档是基于 beta 版本 0.8.2 的当前状态。

要求和兼容性​

¹ClickHouse Connect 已明确测试与所列平台的兼容性。此外，为所有由优秀的 cibuildwheel 项目支持的架构构建了未经测试的二进制轮（带 C 优化）。 最后，由于 ClickHouse Connect 还可以作为纯 Python 运行，因此源安装在任何最近的 Python 安装上都应该适用。

² SQLAlchemy 的支持再次主要限于查询功能。完整的 SQLAlchemy API 不受支持。

³ClickHouse Connect 已针对所有当前支持的 ClickHouse 版本进行测试。由于它使用 HTTP 协议，因此它在大多数其他版本的 ClickHouse 上也应能正确工作，尽管某些高级数据类型可能存在某些不兼容性。

安装​

通过 pip 从 PyPI 安装 ClickHouse Connect：

pip install clickhouse-connect

ClickHouse Connect 也可以从源代码安装：

• git clone GitHub 仓库。
• （可选）运行 pip install cython 以构建和启用 C/Cython 优化。
• cd 到项目根目录并运行 pip install .

支持政策​

ClickHouse Connect 目前处于 beta 版本，只有当前的 beta 版本受到积极支持。在报告任何问题之前，请更新到最新版本。问题应提交到 GitHub 项目。未来版本的 ClickHouse Connect 保证与发布时的积极支持的 ClickHouse 版本兼容（通常是最近三个 stable 和两个最近的 lts 版本）。

基本用法​

收集您的连接详细信息​

要通过 HTTP(S) 连接到 ClickHouse，您需要以下信息：

• 主机和端口：通常，当使用 TLS 时端口为 8443，当不使用 TLS 时端口为 8123。

• 数据库名称：开箱即用时，有一个名为 default 的数据库，请使用您要连接的数据库名称。

• 用户名和密码：开箱即用时，用户名为 default。请使用适合您用例的用户名。

您的 ClickHouse Cloud 服务的详细信息可在 ClickHouse Cloud 控制台中获得。 选择您要连接的服务并点击 连接：

选择 HTTPS，详细信息可在示例的 curl 命令中获得。

如果您使用的是自管理的 ClickHouse，连接详细信息由您的 ClickHouse 管理员设置。

建立连接​

有两个示例展示了如何连接到 ClickHouse：

• 连接到 localhost 上的 ClickHouse 服务器。
• 连接到 ClickHouse Cloud 服务。

使用 ClickHouse Connect 客户端实例连接到 localhost 上的 ClickHouse 服务器：​

import clickhouse_connect

client = clickhouse_connect.get_client(host='localhost', username='default', password='password')

使用 ClickHouse Connect 客户端实例连接到 ClickHouse Cloud 服务：​

import clickhouse_connect

client = clickhouse_connect.get_client(host='HOSTNAME.clickhouse.cloud', port=8443, username='default', password='your password')

与您的数据库交互​

要运行 ClickHouse SQL 命令，请使用客户端的 command 方法：

client.command('CREATE TABLE new_table (key UInt32, value String, metric Float64) ENGINE MergeTree ORDER BY key')

要插入批量数据，请使用客户端的 insert 方法，搭配一个二维数组的行和值：

row1 = [1000, 'String Value 1000', 5.233]
row2 = [2000, 'String Value 2000', -107.04]
data = [row1, row2]
client.insert('new_table', data, column_names=['key', 'value', 'metric'])

要使用 ClickHouse SQL 检索数据，请使用客户端的 query 方法：

result = client.query('SELECT max(key), avg(metric) FROM new_table')
result.result_rows
Out[13]: [(2000, -50.9035)]

ClickHouse Connect 驱动程序 API​

注意： 由于可能的参数数量，建议对大多数 API 方法传递关键字参数，其中大多数都是可选的。

此处未记录的方法不被视为 API的一部分，可能会被移除或更改。

客户端初始化​

clickhouse_connect.driver.client 类提供了 Python 应用程序与 ClickHouse 数据库服务器之间的主要接口。使用 clickhouse_connect.get_client 函数获取 Client 实例，该函数接受以下参数：

连接参数​

HTTPS/TLS 参数​

设置参数​

最后，传递给 get_client 的 settings 参数用于在每个客户端请求中向服务器传递额外的 ClickHouse 设置。请注意，在大多数情况下，具有 readonly=1 访问权限的用户无法更改随查询发送的设置，因此 ClickHouse Connect 将在最终请求中丢弃此类设置并记录警告。以下设置仅适用于 ClickHouse Connect 使用的 HTTP 查询/会话，且未按一般 ClickHouse 设置进行文档化。

有关可以通过每个查询发送的其他 ClickHouse 设置，请参见 ClickHouse 文档。

客户端创建示例​

• 在没有任何参数的情况下，ClickHouse Connect 客户端将连接到 localhost 上的默认 HTTP 端口，使用默认用户且无需密码：

import clickhouse_connect

client = clickhouse_connect.get_client()
client.server_version
Out[2]: '22.10.1.98'

• 连接到安全（https）外部 ClickHouse 服务器

import clickhouse_connect

client = clickhouse_connect.get_client(host='play.clickhouse.com', secure=True, port=443, user='play', password='clickhouse')
client.command('SELECT timezone()')
Out[2]: 'Etc/UTC'

• 连接时使用会话 ID 和其他自定义连接参数及 ClickHouse 设置。

import clickhouse_connect

client = clickhouse_connect.get_client(host='play.clickhouse.com',
                                       user='play',
                                       password='clickhouse',
                                       port=443,
                                       session_id='example_session_1',
                                       connect_timeout=15,
                                       database='github',
                                       settings={'distributed_ddl_task_timeout':300})
client.database
Out[2]: 'github'

常见方法参数​

几个客户端方法使用一个或两个常见的 parameters 和 settings 参数。以下描述这些关键字参数。

参数参数​

ClickHouse Connect 客户端的 query* 和 command 方法接受一个可选的 parameters 关键字参数，用于将 Python 表达式绑定到 ClickHouse 值表达式。可用两种类型的绑定。

服务器端绑定​

ClickHouse 支持 服务器端绑定，用于大多数查询值，其中绑定值与查询分开作为 HTTP 查询参数发送。如果 ClickHouse Connect 检测到形式为 {<name>:<datatype>} 的绑定表达式，将添加相应的查询参数。对于服务器端绑定，parameters 参数应为 Python 字典。

• 使用 Python 字典、DateTime 值和字符串值的服务器端绑定

import datetime

my_date = datetime.datetime(2022, 10, 1, 15, 20, 5)

parameters = {'table': 'my_table', 'v1': my_date, 'v2': "a string with a single quote'"}
client.query('SELECT * FROM {table:Identifier} WHERE date >= {v1:DateTime} AND string ILIKE {v2:String}', parameters=parameters)


# Generates the following query on the server

# SELECT * FROM my_table WHERE date >= '2022-10-01 15:20:05' AND string ILIKE 'a string with a single quote\''

重要 -- 服务器端绑定仅支持（由 ClickHouse 服务器）对于 SELECT 查询。它不适用于 ALTER、DELETE、INSERT 或其他类型的查询。这一点在未来可能会改变，请见 https://github.com/ClickHouse/ClickHouse/issues/42092。

客户端绑定​

ClickHouse Connect 还支持客户端参数绑定，这可以允许在生成模板化 SQL 查询时提供更大的灵活性。对于客户端绑定，parameters 参数应为字典或序列。客户端绑定使用 Python 的 "printf" 风格 字符串格式化进行参数替换。

请注意，与服务器端绑定不同，客户端绑定不适用于数据库标识符，例如数据库、表或列名，因为 Python 风格的格式化无法区分不同类型的字符串，它们需要进行不同的格式化（数据库标识符使用反引号或双引号，数据值使用单引号）。

• 使用 Python 字典、DateTime 值和字符串转义的示例

import datetime

my_date = datetime.datetime(2022, 10, 1, 15, 20, 5)

parameters = {'v1': my_date, 'v2': "a string with a single quote'"}
client.query('SELECT * FROM some_table WHERE date >= %(v1)s AND string ILIKE %(v2)s', parameters=parameters)


# Generates the following query:

# SELECT * FROM some_table WHERE date >= '2022-10-01 15:20:05' AND string ILIKE 'a string with a single quote\''

• 使用 Python 序列（元组）、Float64 和 IPv4Address 的示例

import ipaddress

parameters = (35200.44, ipaddress.IPv4Address(0x443d04fe))
client.query('SELECT * FROM some_table WHERE metric >= %s AND ip_address = %s', parameters=parameters)


# Generates the following query:

# SELECT * FROM some_table WHERE metric >= 35200.44 AND ip_address = '68.61.4.254''

query = 'SELECT {p1:DateTime64(3)}'  # Server side binding with dictionary
parameters={'p1': DT64Param(dt_value)}

query = 'SELECT %s as string, toDateTime64(%s,6) as dateTime' # Client side binding with list 
parameters=['a string', DT64Param(datetime.now())]

query = 'SELECT {p1:DateTime64(3)}, {a1:Array(DateTime(3))}'  # Server side binding with dictionary

parameters={'p1_64': dt_value, 'a1_64': [dt_value1, dt_value2]}

设置参数​

所有关键的 ClickHouse Connect 客户端 “insert” 和 “select” 方法接受一个可选的 settings 关键字参数，以传递 ClickHouse 服务器 用户设置 给包含的 SQL 语句。settings 参数应为字典。每个项目应为 ClickHouse 设置名称及其关联值。请注意，当作为查询参数发送到服务器时，值将被转换为字符串。

与客户端级设置一样，ClickHouse Connect 将丢弃服务器标记为 readonly=1 的任何设置，并附带日志消息。仅适用于通过 ClickHouse HTTP 接口进行查询的设置始终有效。这些设置在 get_client API 下进行描述。

使用 ClickHouse 设置的示例：

settings = {'merge_tree_min_rows_for_concurrent_read': 65535,
            'session_id': 'session_1234',
            'use_skip_indexes': False}
client.query("SELECT event_type, sum(timeout) FROM event_errors WHERE event_time > '2022-08-01'", settings=settings)

客户端 command 方法​

使用 Client.command 方法将 SQL 查询发送到 ClickHouse 服务器，这些查询通常不返回数据或返回单个原始值或数组值，而不是完整的数据集。此方法接受以下参数：

• _command_ 可用于 DDL 语句。如果 SQL “命令”没有返回数据，则返回一个“查询摘要”字典。该字典封装了 ClickHouse X-ClickHouse-Summary 和 X-ClickHouse-Query-Id 头，包括键/值对 written_rows、written_bytes 和 query_id。

client.command('CREATE TABLE test_command (col_1 String, col_2 DateTime) Engine MergeTree ORDER BY tuple()')
client.command('SHOW CREATE TABLE test_command')
Out[6]: 'CREATE TABLE default.test_command\\n(\\n    `col_1` String,\\n    `col_2` DateTime\\n)\\nENGINE = MergeTree\\nORDER BY tuple()\\nSETTINGS index_granularity = 8192'

• _command_ 还可以用于仅返回单行的简单查询

result = client.command('SELECT count() FROM system.tables')
result
Out[7]: 110

客户端 query 方法​

Client.query 方法是从 ClickHouse 服务器检索单个 “批次” 数据集的主要方式。它利用 HTTP 的原生 ClickHouse 格式高效地传输大型数据集（最多约一百万行）。此方法接受以下参数。

The QueryResult object​

基础 query 方法返回一个 QueryResult 对象，具有以下公共属性：

• result_rows -- 以行序列的形式返回的数据矩阵，每行元素是列值的序列。
• result_columns -- 以列序列的形式返回的数据矩阵，每列元素是该列的行值序列。
• column_names -- 表示 result_set 中列名的字符串元组。
• column_types -- 表示 result_columns 中每列的 ClickHouse 数据类型的 ClickHouseType 实例的元组。
• query_id -- ClickHouse 的 query_id（对于检查 system.query_log 表中的查询非常有用）。
• summary -- 由 X-ClickHouse-Summary HTTP 响应头返回的任意数据。
• first_item -- 方便属性，用于将响应的第一行作为字典检索（键为列名）。
• first_row -- 返回结果的第一行的方便属性。
• column_block_stream -- 以列为导向格式生成查询结果的生成器。该属性不应被直接引用（见下文）。
• row_block_stream -- 以行为导向格式生成查询结果的生成器。该属性不应被直接引用（见下文）。
• rows_stream -- 生成每次调用返回一行查询结果的生成器。该属性不应被直接引用（见下文）。
• summary -- 如在 command 方法下所述，由 ClickHouse 返回的汇总信息字典。

*_stream 属性返回一个 Python 上下文，可用作返回数据的迭代器。它们只能通过 Client 的 *_stream 方法间接访问。

使用 StreamContext 对象流式查询结果的完整细节在 Advanced Queries (Streaming Queries) 中进行了概述。

Consuming query results with NumPy, Pandas or Arrow​

主 query 方法有三种专用版本：

• query_np -- 此版本返回一个 NumPy 数组，而不是 ClickHouse Connect QueryResult。
• query_df -- 此版本返回一个 Pandas DataFrame，而不是 ClickHouse Connect QueryResult。
• query_arrow -- 此版本返回一个 PyArrow 表。它直接使用 ClickHouse 的 Arrow 格式，因此仅接受与主 query 方法共同的三个参数： query，parameters 和 settings。此外，还有一个额外参数 use_strings，用于确定 Arrow 表是否将 ClickHouse 的字符串类型渲染为字符串（如果为 True）或字节（如果为 False）。

Client streaming query methods​

ClickHouse Connect 客户端提供多种方法以流的形式检索数据（实现为 Python 生成器）：

• query_column_block_stream -- 以块形式将查询数据返回为列的序列，使用原生 Python 对象。
• query_row_block_stream -- 将查询数据作为行块返回，使用原生 Python 对象。
• query_rows_stream -- 将查询数据返回为行的序列，使用原生 Python 对象。
• query_np_stream -- 将每个 ClickHouse 查询数据块作为 NumPy 数组返回。
• query_df_stream -- 将每个 ClickHouse 查询数据块返回为 Pandas DataFrame。
• query_arrow_stream -- 将查询数据以 PyArrow RecordBlocks 返回。

这些方法中的每一个都返回一个 ContextStream 对象，必须通过 with 语句打开以开始消费流。有关详细信息和示例，请参见 Advanced Queries (Streaming Queries)。

Client insert method​

对于将多个记录插入 ClickHouse 的常见用例，有 Client.insert 方法。它接受以下参数：

该方法返回一个“查询摘要”字典，如 "command" 方法下所述。如果因为任何原因插入失败，将引发异常。

主 insert 方法有两个专用版本：

• insert_df -- 此方法的第二个参数要求一个 Pandas DataFrame 实例作为 df 参数，而不是 Python 的序列的序列的 data 参数。ClickHouse Connect 会自动将 DataFrame 处理为列导向数据源，因此 column_oriented 参数不需要也不可用。
• insert_arrow -- 此方法需要一个 arrow_table，而不是 Python 的序列的序列的 data 参数。ClickHouse Connect 将箭头表不做修改地传递给 ClickHouse 服务器进行处理，因此除了 table 和 arrow_table 外，只有 database 和 settings 参数可用。

注意： NumPy 数组是有效的序列的序列，可以用作主 insert 方法的 data 参数，因此不需要专用方法。

File Inserts​

clickhouse_connect.driver.tools 包含 insert_file 方法，允许直接从文件系统插入数据到现有的 ClickHouse 表。解析工作委托给 ClickHouse 服务器。insert_file 接受以下参数：

对于具有不一致数据或日期/时间值格式异常的文件，适用于数据导入的设置（例如 input_format_allow_errors_num）被此方法识别。

import clickhouse_connect
from clickhouse_connect.driver.tools import insert_file

client = clickhouse_connect.get_client()
insert_file(client, 'example_table', 'my_data.csv',
            settings={'input_format_allow_errors_ratio': .2,
                      'input_format_allow_errors_num': 5})

Saving query results as files​

您可以使用 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'
    fmt = 'CSVWithNames'  # or CSV, or CSVWithNamesAndTypes, or TabSeparated, etc.
    stream = client.raw_stream(query=query, fmt=fmt)
    with open("output.csv", "wb") as f:
        for chunk in stream:
            f.write(chunk)

上面的代码生成一个名为 output.csv 的文件，内容如下：

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

同样，您可以以 TabSeparated 和其他格式保存数据。有关所有可用格式选项的概述，请参见 Formats for Input and Output Data。

Raw API​

对于不需要 ClickHouse 数据与本地或第三方数据类型和结构之间转换的用例，ClickHouse Connect 客户端提供了两个用于直接使用 ClickHouse 连接的方法。

Client raw_query Method​

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

处理返回的 bytes 对象是调用方的责任。注意，Client.query_arrow 只是一个围绕此方法的薄包装，使用 ClickHouse 的 Arrow 输出格式。

Client raw_stream Method​

Client.raw_stream 方法与 raw_query 方法具有相同的 API，但返回一个 io.IOBase 对象，可用作 bytes 对象的生成器/流源。它目前由 query_arrow_stream 方法使用。

Client raw_insert Method​

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

确保 insert_block 是指定格式并使用特定压缩方法是调用方的责任。ClickHouse Connect 使用这些原始插入进行文件上传和 PyArrow 表，并将解析委托给 ClickHouse 服务器。

Utility classes and functions​

以下类和函数也被视为“公共” clickhouse-connect API 的一部分，并且与上述文档中的类和方法一样，在小版本更新中是稳定的。对这些类和函数的重大更改只会在小版本（而不是补丁）发布时发生，并且在至少一个小版本发布中将以弃用状态提供。

Exceptions​

所有自定义异常（包括在 DB API 2.0 规范中定义的异常）都在 clickhouse_connect.driver.exceptions 模块中定义。驱动程序检测到的异常将使用这些类型中的一种。

Clickhouse SQL utilities​

clickhouse_connect.driver.binding 模块中的函数和 DT64Param 类可用于正确构建和转义 ClickHouse SQL 查询。同样，clickhouse_connect.driver.parser 模块中的函数可用于解析 ClickHouse 数据类型名称。

Multithreaded, multiprocess, and async/event driven use cases​

ClickHouse Connect 在多线程、多进程和事件循环驱动/异步应用程序中表现良好。所有查询和插入处理都发生在单个线程内，因此操作通常是线程安全的。（低级别的某些操作的并行处理可能是未来的增强功能，以克服单线程的性能损失， 但是即使在那种情况下也会保持线程安全）。

由于每个查询或插入执行都会在其自己的 QueryContext 或 InsertContext 对象中维护状态，因此这些辅助对象不是线程安全的，且不应在多个处理流之间共享。有关上下文对象的额外讨论，请参见以下部分。

此外，在具有两个或多个 "正在进行" 的查询和/或插入的应用程序中，需要牢记两个进一步的考虑因素。第一个是与查询/插入关联的 ClickHouse “会话”，第二个是 ClickHouse Connect 客户端实例使用的 HTTP 连接池。

Asyncclient wrapper​

自 0.7.16 起，ClickHouse Connect 提供了一个异步包装，覆盖常规的 Client，因此可以在 asyncio 环境中使用客户端。

要获取 AsyncClient 的实例，可以使用 get_async_client 工厂函数，接受与标准 get_client 相同的参数：

import asyncio

import clickhouse_connect

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

asyncio.run(main())

AsyncClient 具有与标准 Client 相同的方法和相同的参数，但在适用时是协程。内部，Client 中执行 I/O 操作的方法被包装在一个 run_in_executor 调用中。

使用 AsyncClient 包装器时，多线程性能将提高，因为在等待 I/O 操作完成时将释放执行线程和 GIL。

注意：与常规 Client 不同，AsyncClient 默认强制 autogenerate_session_id 为 False。

另请参见：run_async example。

Managing ClickHouse Session Ids​

每个 ClickHouse 查询都在 ClickHouse “会话”的上下文中发生。会话目前有两个用途：

• 将特定的 ClickHouse 设置与多个查询关联（参见 user settings）。使用 ClickHouse 的 SET 命令更改用户会话范围的设置。
• 跟踪 temporary tables.

默认情况下，使用 ClickHouse Connect 客户端实例执行的每个查询都使用相同的会话 ID，以启用该会话功能。即，当使用单一 ClickHouse 客户端时，SET 语句和临时表的工作都如预期般正常。然而，设计上 ClickHouse 服务器不允许在同一会话内进行并发查询。因此，对于将执行并发查询的 ClickHouse Connect 应用程序，有两个选择。

• 为将具有自己会话 ID 的每个执行线程（线程、进程或事件处理程序）创建一个单独的 Client 实例。这通常是最佳方法，因为它保留了每个客户端的会话状态。
• 为每个查询使用唯一的会话 ID。在临时表或共享会话设置不需要的情况下，这可避免并发会话问题。（创建客户端时，也可以提供共享设置，但这些设置与会话无关，而是与每个请求一起发送）。唯一的 session_id 可以添加到每个请求的 settings 字典中，或者您可以禁用 autogenerate_session_id 通用设置：

from clickhouse_connect import common

common.set_setting('autogenerate_session_id', False)  # This should always be set before creating a client
client = clickhouse_connect.get_client(host='somehost.com', user='dbuser', password=1234)

在这种情况下，ClickHouse Connect 将不发送任何会话 ID，ClickHouse 服务器将生成随机会话 ID。再次说明，临时表和会话级别设置将不可用。

Customizing the HTTP connection pool​

ClickHouse Connect 使用 urllib3 连接池处理与服务器的基础 HTTP 连接。默认情况下，所有客户端实例共享相同的连接池，这对于大多数用例足够。此默认池维护最多 8 个 HTTP Keep Alive 连接到应用程序使用的每个 ClickHouse 服务器。

对于大型多线程应用程序，可能需要单独的连接池。可以将自定义连接池作为 pool_mgr 关键字参数提供给主 clickhouse_connect.get_client 函数：

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)

如上例所示，客户端可以共享一个池管理器，或为每个客户端创建一个单独的池管理器。有关创建 PoolManager 时可用选项的更多详细信息，请参见 urllib3 documentation。

Querying data with ClickHouse Connect: Advanced usage​

QueryContexts​

ClickHouse Connect 在 QueryContext 中执行标准查询。QueryContext 包含用于构建针对 ClickHouse 数据库查询的关键结构，以及用于将结果处理为 QueryResult 或其他响应数据结构的配置。包括查询本身、参数、设置、读取格式和其他属性。

可以使用客户端的 create_query_context 方法获取 QueryContext。此方法接受与核心查询方法相同的参数。然后可以将此查询上下文作为 context 关键字参数传递给 query、query_df 或 query_np 方法，而不需要提供任何或全部其他参数。请注意，为方法调用指定的其他参数将覆盖 QueryContext 的任何属性。

QueryContext 的最清晰用例是在发送相同查询但不同绑定参数值时。可以通过调用 QueryContext.set_parameters 方法并传递一个字典来更新所有参数值，或通过调用 QueryContext.set_parameter 方法及所需的 key、value 对来更新单个值。

client.create_query_context(query='SELECT value1, value2 FROM data_table WHERE key = {k:Int32}',
                            parameters={'k': 2},
                            column_oriented=True)
result = client.query(context=qc)
assert result.result_set[1][0] == 'second_value2'
qc.set_parameter('k', 1)
result = test_client.query(context=qc)
assert result.result_set[1][0] == 'first_value2'

请注意，QueryContexts 不是线程安全的，但可以通过调用 QueryContext.updated_copy 方法在多线程环境中获得副本。

Streaming queries​

Data blocks​

ClickHouse Connect 将来自主 query 方法的所有数据处理为从 ClickHouse 服务器接收的块流。这些块以自定义的 "Native" 格式在 ClickHouse 之间传输。“块”只是相同数据类型的二进制数据列的序列。（作为列式数据库，ClickHouse 以类似的形式存储这些数据。）从查询返回的块的大小由两个用户设置控制，可以在多个级别设置（用户配置文件、用户、会话或查询）。它们是：

• max_block_size -- 块大小（以行为单位）的限制。默认值为 65536。
• preferred_block_size_bytes -- 块大小（以字节为单位）的软限制。默认值为 1,000,0000。

无论 preferred_block_size_setting 如何，每个块的行数永远不会超过 max_block_size。具体查询返回的实际块可以是任意大小。例如，覆盖多个分片的分布式表的查询可能包含直接从每个分片检索的小块。

使用客户端的 query_*_stream 方法时，结果按块返回。ClickHouse Connect 仅加载一个块。这样可以处理大量数据，而无需将大量结果集全部加载到内存中。请注意，应用程序应准备处理任意数量的块，并且每个块的确切大小无法控制。

HTTP data buffer for slow processing​

由于 HTTP 协议的限制，如果块的处理速度显著慢于 ClickHouse 服务器流式传输数据的速度，ClickHouse 服务器将关闭连接，导致处理线程中抛出异常。通过使用通用的 http_buffer_size 设置增加 HTTP 流式缓冲区的缓冲区大小（默认值为 10 兆字节），可以减轻一些。 在这种情况下，如果应用程序有足够的内存，使用大的 http_buffer_size 值应该是安全的。如果使用 lz4 或 zstd 压缩，则缓冲区中的数据会被压缩，因此使用这些压缩类型将增加可用的整体缓冲区。

StreamContexts​

每个 query_*_stream 方法（如 query_row_block_stream）返回一个 ClickHouse StreamContext 对象，它是一个结合的 Python 上下文/生成器。这是基本的用法：

with client.query_row_block_stream('SELECT pickup, dropoff, pickup_longitude, pickup_latitude FROM taxi_trips') as stream:
    for block in stream:
        for row in block:
            <do something with each row of Python trip data>

请注意，尝试在没有 with 语句的情况下使用 StreamContext 将引发错误。使用 Python 上下文可确保即使没有消耗所有数据和/或在处理过程中出现异常，流（在这种情况下是流式的 HTTP 响应）也会被正确关闭。此外，StreamContexts 只能使用一次来消费流。在退出后尝试使用 StreamContext 将产生 StreamClosedError。

可以使用 StreamContext 的 source 属性访问父 QueryResult 对象，其中包含列名和类型。

Stream types​

query_column_block_stream 方法将块作为存储为原生 Python 数据类型的列数据序列返回。使用上述的 taxi_trips 查询，返回的数据将是一个列表，其中每个元素是一个列表（或元组），包含关联列的所有数据。因此，block[0] 将是一个仅包含字符串的元组。列导向格式最常用于对列中所有值进行聚合操作，例如总计车费。

query_row_block_stream 方法将块作为行序列返回，类似于传统关系数据库。对于 taxi trips，返回的数据将是一个列表，其中每个元素是另一个列表，表示一行数据。因此，block[0] 将包含第一辆出租车行程的所有字段（按顺序），block[1] 将包含第二辆出租车行程的所有字段，以此类推。行导向的结果通常用于显示或转换过程。

query_row_stream 是一个方便的方法，在迭代流时会自动移动到下一个块。否则，它与 query_row_block_stream 完全相同。

query_np_stream 方法将每个块作为二维 NumPy 数组返回。在内部，NumPy 数组（通常）按列存储，因此不需要单独的行或列方法。NumPy 数组的“形状”将表示为（列，行）。NumPy 库提供了许多操作 NumPy 数组的方法。请注意，如果查询中的所有列共享相同的 NumPy 数据类型，则返回的 NumPy 数组也只会有一个数据类型，并且可以在不实际更改其内部结构的情况下重新塑造/旋转。

query_df_stream 方法将每个 ClickHouse 块返回为二维 Pandas DataFrame。以下是一个示例，显示 StreamContext 对象可以以延迟的方式作为上下文使用（但只能使用一次）。

最后，query_arrow_stream 方法返回 ClickHouse ArrowStream 格式的结果，包装在 StreamContext 中为 pyarrow.ipc.RecordBatchStreamReader。流的每次迭代返回 PyArrow RecordBlock。

df_stream = client.query_df_stream('SELECT * FROM hits')
column_names = df_stream.source.column_names
with df_stream:
    for df in df_stream:
        <do something with the pandas DataFrame>

Read formats​

读取格式控制从客户端的 query、query_np 和 query_df 方法返回的值的数据类型。（raw_query 和 query_arrow 不会修改来自 ClickHouse 的输入数据，因此格式控制不适用。）例如，如果将 UUID 的读取格式从默认的 native 格式更改为可选的 string 格式，则 ClickHouse 的 UUID 列查询将返回字符串值（使用标准的 8-4-4-4-12 RFC 1422 格式），而不是 Python UUID 对象。

任何格式化函数的“数据类型”参数可以包括通配符。格式是一个单一的小写字符串。

读取格式可以在多个级别设置：

• 全局设置，使用 clickhouse_connect.datatypes.format 包中定义的方法。这将控制配置数据类型的格式，以适用于所有查询。

from clickhouse_connect.datatypes.format import set_read_format


# Return both IPv6 and IPv4 values as strings
set_read_format('IPv*', 'string')


# Return all Date types as the underlying epoch second or epoch day
set_read_format('Date*', 'int')

• 针对整个查询，使用可选的 query_formats 字典参数。在这种情况下，任何列（或子列）的指定数据类型会使用配置的格式。

# Return any UUID column as a string
client.query('SELECT user_id, user_uuid, device_uuid from users', query_formats={'UUID': 'string'})

• 针对特定列中的值，使用可选的 column_formats 字典参数。键是 ClickHouse 返回的列名，以及数据列的格式或第二级 “格式” 字典，ClickHouse 类型名称和查询格式值。此二级字典可用于嵌套列类型，例如元组或图。

# Return IPv6 values in the `dev_address` column as strings
client.query('SELECT device_id, dev_address, gw_address from devices', column_formats={'dev_address':'string'})

读取格式选项（Python 类型）​

外部数据​

ClickHouse 查询可以接受任何 ClickHouse 格式的外部数据。此二进制数据与查询字符串一起发送，以用于处理数据。有关外部数据特性的详细信息请参见 这里。客户端 query* 方法接受可选的 external_data 参数以利用此功能。external_data 参数的值应该是 clickhouse_connect.driver.external.ExternalData 对象。该对象的构造函数接受以下参数：

要发送包含“电影”数据的外部 CSV 文件的查询，并将该数据与 ClickHouse 服务器上已存在的 directors 表结合：

import clickhouse_connect
from clickhouse_connect.driver.external import ExternalData

client = clickhouse_connect.get_client()
ext_data = ExternalData(file_path='/data/movies.csv',
                        fmt='CSV',
                        structure=['movie String', 'year UInt16', 'rating Decimal32(3)', 'director String'])
result = client.query('SELECT name, avg(rating) FROM directors INNER JOIN movies ON directors.name = movies.director GROUP BY directors.name',
                      external_data=ext_data).result_rows

可以使用 add_file 方法将其他外部数据文件添加到初始 ExternalData 对象，该方法接受与构造函数相同的参数。对于 HTTP，所有外部数据作为 multi-part/form-data 文件上传的一部分进行传输。

时区​

有多种机制可将时区应用于 ClickHouse DateTime 和 DateTime64 值。内部的 ClickHouse 服务器始终将任何 DateTime 或 DateTime64 对象存储为表示自纪元以来的无时区的数字，即 1970-01-01 00:00:00 UTC 时间。对于 DateTime64 值，表示可以是自纪元以来的毫秒、微秒或纳秒，具体取决于精度。因此，任何时区信息的应用总是在客户端进行。请注意，这涉及到有意义的额外计算，因此在性能关键的应用程序中，建议将 DateTime 类型视为纪元时间戳，除用户显示和转换之外（例如，Pandas 时间戳总是表示纪元纳秒的 64 位整数，以提高性能）。

在查询中使用时区感知数据类型时 - 特别是 Python 的 datetime.datetime 对象 -- clickhouse-connect 使用以下优先规则来应用客户端时区：

1. 如果为查询指定了查询方法参数 client_tzs，则应用特定列的时区。
2. 如果 ClickHouse 列具有时区元数据（即这是类型 DateTime64(3, 'America/Denver')），则应用 ClickHouse 列的时区。（请注意，该时区元数据在 ClickHouse 版本 23.2 之前对 clickhouse-connect 不可用）
3. 如果为查询指定了查询方法参数 query_tz，则应用“查询时区”。
4. 如果在查询或会话中应用了时区设置，则应用该时区。（此功能尚未在 ClickHouse 服务器中发布）
5. 最后，如果客户端 apply_server_timezone 参数已设置为 True（默认值），则应用 ClickHouse 服务器时区。

请注意，如果根据这些规则应用的时区为 UTC，clickhouse-connect 将 始终 返回一个无时区的 Python datetime.datetime 对象。如果需要，可以通过应用程序代码向该无时区对象添加额外的时区信息。

使用 ClickHouse Connect 插入数据：高级用法​

InsertContexts​

ClickHouse Connect 在 InsertContext 内执行所有插入。InsertContext 包含发送到客户端 insert 方法的所有值。此外，当最初构造 InsertContext 时，ClickHouse Connect 检索执行高效本机格式插入所需的插入列的数据类型。通过重用 InsertContext 进行多次插入，可以避免这个“预查询”，并且插入执行得更快、更高效。

可以使用客户端 create_insert_context 方法获取 InsertContext。该方法接受与 insert 函数相同的参数。请注意，仅应修改 InsertContexts 的 data 属性以进行重用。这与其提供一个可重复使用对象以将新数据重复插入同一表的预期用途一致。

test_data = [[1, 'v1', 'v2'], [2, 'v3', 'v4']]
ic = test_client.create_insert_context(table='test_table', data=test_data)
client.insert(context=ic)
assert client.command('SELECT count() FROM test_table') == 2
new_data = [[3, 'v5', 'v6'], [4, 'v7', 'v8']]
ic.data = new_data
client.insert(context=ic)
qr = test_client.query('SELECT * FROM test_table ORDER BY key DESC')
assert qr.row_count == 4
assert qr[0][0] == 4

InsertContexts 包含在插入过程中更新的可变状态，因此它们不是线程安全的。

写入格式​

目前，写入格式仅针对有限数量的类型实现。在大多数情况下，ClickHouse Connect 将尝试通过检查第一个（非空）数据值的类型来自动确定列的正确写入格式。例如，如果插入到 DateTime 列，而且该列的第一个插入值是 Python 整数，ClickHouse Connect 将直接插入该整数值，假设它实际上是一个纪元秒。

在大多数情况下，覆盖数据类型的写入格式不必要，但 clickhouse_connect.datatypes.format 包中的相关方法可以在全局级别使用。

写入格式选项​

附加选项​

ClickHouse Connect 为高级用例提供了一些附加选项

全局设置​

有少量设置可以全局控制 ClickHouse Connect 的行为。它们可以从顶层 common 包访问：

from clickhouse_connect import common

common.set_setting('autogenerate_session_id', False)
common.get_setting('invalid_setting_action')
'drop'

当前定义了十个全局设置：

压缩​

ClickHouse Connect 支持 lz4、zstd、brotli 和 gzip 压缩，适用于查询结果和插入。始终记住，使用压缩通常涉及在网络带宽/传输速度与 CPU 使用之间的权衡（无论是在客户端还是服务器上）。

要接收压缩数据，ClickHouse 服务器 enable_http_compression 必须设置为 1，或者用户必须有权限在“每查询”基础上更改该设置。

压缩由调用 clickhouse_connect.get_client 工厂方法时的 compress 参数控制。默认情况下，compress 设置为 True，这将触发默认的压缩设置。对于使用 query、query_np 和 query_df 客户端方法执行的查询，ClickHouse Connect 将添加 Accept-Encoding 头，带有 lz4、zstd、br（如果安装了 brotli 库则为 brotli）、gzip 和 deflate 编码，在使用 query 客户端方法执行的查询中（间接地，query_np 和 query_df）。对于大多数请求，ClickHouse 服务器将以 zstd 压缩负载返回。对于插入，默认情况下，ClickHouse Connect 将使用 lz4 压缩插入块，并发送 Content-Encoding: lz4 HTTP 头。

get_client 的 compress 参数也可以设置为特定的压缩方法之一，lz4、zstd、br 或 gzip。然后将对插入和查询结果（如果 ClickHouse 服务器支持）都使用该方法。所需的 zstd 和 lz4 压缩库现在默认与 ClickHouse Connect 一起安装。如果指定了 br/brotli，则必须单独安装 brotli 库。

请注意，raw* 客户端方法不使用客户端配置指定的压缩。

我们还建议不要使用 gzip 压缩，因为它在压缩和解压缩数据时显著比其他选项要慢。

HTTP 代理支持​

ClickHouse Connect 使用 urllib3 库添加基本的 HTTP 代理支持。它识别标准的 HTTP_PROXY 和 HTTPS_PROXY 环境变量。请注意，使用这些环境变量将适用于任何使用 clickhouse_connect.get_client 方法创建的客户端。或者，可以使用 http_proxy 或 https_proxy 参数来为每个客户端进行配置。有关 HTTP 代理支持实现的详细信息，请参阅 urllib3 文档。

要使用 Socks 代理，您可以将 urllib3 的 SOCKSProxyManager 作为 pool_mgr 参数发送到 get_client。请注意，这需要安装 PySocks 库，无论是直接安装还是通过 urllib3 依赖的 [socks] 选项。

“旧” JSON 数据类型​

实验性 Object（或 Object('json')）数据类型已被弃用，并应避免在生产环境中使用。ClickHouse Connect 继续为该数据类型提供有限支持，以实现向后兼容性。请注意，此支持不包括期望返回“顶层”或“父级” JSON 值作为字典或等效内容的查询，这样的查询将导致异常。

“新” Variant/Dynamic/JSON 数据类型（实验特性）​

从 0.8.0 版本开始，clickhouse-connect 提供对新（也是实验性）ClickHouse 类型 Variant、Dynamic 和 JSON 的实验性支持。

使用备注​

• JSON 数据可以作为 Python 字典或包含 JSON 对象 {} 的 JSON 字符串插入。其他形式的 JSON 数据不被支持。
• 使用子列/路径的查询将返回子列的类型。
• 有关其他使用备注，请参阅主 ClickHouse 文档。

已知限制​

• 必须在使用之前在 ClickHouse 设置中启用每种类型。
• “新” JSON 类型从 ClickHouse 24.8 发布开始可用。
• 由于内部格式的变化，clickhouse-connect 仅与 ClickHouse 24.7 发布后的 Variant 类型兼容。
• 返回的 JSON 对象仅返回 max_dynamic_paths 数量的元素（默认为 1024）。这将在未来的发布中修复。
• 插入到 Dynamic 列的内容将始终为 Python 值的字符串表示。此问题将在未来的发布中修复，一旦 https://github.com/ClickHouse/ClickHouse/issues/70395 得到解决。
• 对新类型的实现尚未在 C 代码中优化，因此性能可能比简单、成熟的数据类型要慢一些。