高级插入 | ClickHouse Docs

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

InsertContexts

ClickHouse Connect 的 Native-format 插入操作，以及 insert 和 insert_df 方法，都在 InsertContext 中执行。insert_arrow、insert_df_arrow 和 raw_insert 方法会直接发送其载荷，而不会使用 InsertContext。InsertContext 包含传递给客户端 insert 方法的所有参数值。此外，在初次构造 InsertContext 时，ClickHouse Connect 还会获取插入列的数据类型，以便高效地进行 Native format 插入。复用同一个 InsertContext 执行多次插入时，就可以避免这类“预查询”，从而让插入更快、更高效。

可以使用客户端的 create_insert_context 方法获取 InsertContext。该方法接受的参数与 insert 函数相同，但 context 本身除外。请注意，复用 InsertContext 时，只应修改其 data 属性。这也符合它的设计目的：为向同一张表重复插入新数据提供一个可复用的对象。

test_data = [[13, "v1", "v2"], [79, "v3", "v4"]]
ic = 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 = [[101, "v5", "v6"], [113, "v7", "v8"]]
ic.data = new_data
client.insert(context=ic)
qr = client.query("SELECT * FROM test_table ORDER BY key DESC")
assert qr.row_count == 4
assert qr.first_row[0] == 113

InsertContexts 包含会在 insert 过程中更新的可变状态，因此不具备线程安全性。

写入格式

只有少数类型实现了写入格式。在大多数情况下，ClickHouse Connect 会根据列中第一个非 NULL 的数据值自动判断正确的写入格式。例如，当 DateTime 列的第一个值是整数时，client 会将其视为纪元秒。

通常无需覆盖写入格式，但 clickhouse_connect.datatypes.format 中的方法可以在全局范围内进行设置。像 Array、Nullable 和 LowCardinality 这样的容器包装器会保留其元素类型的格式行为。

写入格式选项

ClickHouse 类型	原生 Python 类型	写入格式	注释
Int[8-64], UInt[8-32]	int
UInt64	int
[U]Int[128,256]	int
BFloat16	float
Float32	float
Float64	float
Decimal	decimal.Decimal
String	str or bytes		同一列中的值必须始终统一为文本或字节类型。
FixedString	bytes	string	字符串值会以零字节填充。空字节会写入为全零字节。
Enum[8,16]	str or int		可将标签作为字符串插入，或插入其底层整数值。
Date	datetime.date	int	整数值会被解释为自 1970-01-01 起的天数。
Date32	datetime.date	int	整数值会被解释为有符号的天数偏移。
DateTime	datetime.datetime	int	整数值会被解释为纪元秒。
DateTime64	datetime.datetime	int	整数值会被解释为按列精度表示的 tick。
Time	datetime.timedelta	int, string, time	整数值会被解释为秒。
Time64	datetime.timedelta	int, string, time	整数值会被解释为按列精度表示的 tick。
IPv4	`ipaddress.IPv4Address`	string	格式正确的字符串可作为 IPv4 地址插入
IPv6	`ipaddress.IPv6Address`	string	格式正确的字符串可作为 IPv6 地址插入
Tuple	dict or tuple
Map	dict
Nested	Sequence[dict]
UUID	uuid.UUID	string	格式正确的字符串可作为 ClickHouse UUID 插入
JSON	dict	string	支持字典和 JSON 对象字符串。旧版 `Object('json')` 类型不受支持。
Variant	object		值使用原生成员序列化。在 Python 类型存在歧义时，请使用 `clickhouse_connect.datatypes.dynamic.typed_variant`。
Dynamic	object		当前值通过其 String 表示形式插入。
QBit	Sequence[float]		安装 NumPy 后，会自动使用它来更快地进行比特转置。

专用插入方法

ClickHouse Connect 为常见数据格式提供了专用的插入方法：

insert_df – 将 Pandas DataFrame 作为列式 Native 数据插入。它还支持显式指定列名/类型，或使用可复用的 InsertContext。
insert_arrow – 使用 ClickHouse Arrow 输入格式插入 PyArrow Table。
insert_df_arrow – 插入基于 Arrow 的 Pandas DataFrame 或 Polars DataFrame。Pandas 的所有列都必须使用基于 Arrow 的 dtype。

这三种方法都接受 database、settings 和按请求指定的 HTTP transport_settings。

插入 Pandas DataFrame

import clickhouse_connect
import pandas as pd

client = clickhouse_connect.get_client()

df = pd.DataFrame({
    "id": [13, 79],
    "name": ["user_1", "user_2"],
    "age": [25, 30],
})

client.insert_df("users", df)

PyArrow Table 插入操作

import clickhouse_connect
import pyarrow as pa

client = clickhouse_connect.get_client()

arrow_table = pa.table({
    "id": [13, 79],
    "name": ["user_1", "user_2"],
    "age": [25, 30],
})

client.insert_arrow("users", arrow_table)

基于 Arrow 的 DataFrame 插入 (pandas 2.x)

import clickhouse_connect
import pandas as pd

client = clickhouse_connect.get_client()

# Convert to Arrow-backed dtypes for better performance
df = pd.DataFrame({
    "id": [13, 79],
    "name": ["user_1", "user_2"],
    "age": [25, 30],
}).convert_dtypes(dtype_backend="pyarrow")

client.insert_df_arrow("users", df)

根据 PyArrow schema 创建表

create_table_from_arrow_schema 会根据常见的标量 Arrow field 构建 CREATE TABLE 语句。该映射涵盖有符号和无符号整数、浮点值、布尔值、字符串、日期和时间戳。它会刻意创建不可为 NULL 的 ClickHouse 列，并在遇到不受支持的 Arrow type 时引发 TypeError，因此请在执行前先检查生成的 DDL。

import clickhouse_connect
import pyarrow as pa

from clickhouse_connect.driver.ddl import create_table_from_arrow_schema

client = clickhouse_connect.get_client()
schema = pa.schema(
    [
        ("id", pa.uint32()),
        ("name", pa.string()),
        ("event_time", pa.timestamp("ms", tz="UTC")),
    ]
)
ddl = create_table_from_arrow_schema(
    table_name="arrow_events",
    schema=schema,
    engine="MergeTree",
    engine_params={"ORDER BY": "id"},
)
client.command(ddl)

时区

将 Python datetime 对象插入 DateTime 或 DateTime64 列时，ClickHouse Connect 会将其转换为自纪元开始计数的值。

带时区信息的 datetime 对象

带时区信息的对象会保留其所表示的时刻。源时区无需与 ClickHouse 列中声明的时区一致。

from datetime import datetime, timezone
from zoneinfo import ZoneInfo

client.command("CREATE TABLE events (event_time DateTime) ENGINE Memory")

data = [
    [datetime(2023, 6, 15, 10, 30, tzinfo=timezone.utc)],
    [datetime(2023, 6, 15, 10, 30, tzinfo=ZoneInfo("America/Denver"))],
    [datetime(2023, 6, 15, 10, 30, tzinfo=ZoneInfo("Asia/Tokyo"))],
]

client.insert("events", data, column_names=["event_time"])
results = client.query(
    "SELECT event_time FROM events ORDER BY event_time",
    query_tz="UTC",
    tz_mode="aware",
)
assert [row[0].hour for row in results.result_rows] == [1, 10, 16]

不带时区的 datetime 对象

全局 naive_datetime_insert 设置用于控制插入不带时区的原生 Python datetime 对象。它也适用于 DateTime64 列接受的不带时区的 ISO 字符串。

"local" 是 1.x 中的默认值。调用 .timestamp() 时，Python 会根据进程时区解释该值，从而保持现有行为。
"server" 会将该值解释为 DateTime 或 DateTime64 列所声明时区中的墙钟时间。如果该列未指定时区，则使用客户端连接时获取的服务器时区。

请在插入前设置该选项。序列化每个包含 Python datetime 对象或 DateTime64 ISO 字符串的原生插入列时，都会读取此选项，因此更改会应用于现有客户端和可复用的插入上下文。

from datetime import datetime

from clickhouse_connect import common

common.set_setting("naive_datetime_insert", "server")

naive_time = datetime(2023, 6, 15, 10, 30)
client.insert("events", [[naive_time]], column_names=["event_time"])

使用 "server" 时，ClickHouse Connect 会先附加目标 tzinfo，再将值转换为纪元时间。对于 IANA 时区，它遵循标准库处理夏令时切换的规则。秋季重叠时段使用 datetime 的 fold 值。默认的 fold=0 选择切换前的偏移量，而 fold=1 选择切换后的偏移量。春季间隙采用相同的偏移量选择方式，不会被拒绝或归一化。

不存在的春季间隙墙钟时间可能无法通过墙钟模式查询参数往返转换，因为 ClickHouse 的文本解析可能会选择不同的偏移量。当具体时刻很重要时，请使用带时区信息的 datetime 或有效的墙钟时间。

该选项仅适用于原生 Python datetime 对象的插入，以及 DateTime64 接受的不带时区的 ISO 字符串。不带时区的 datetime64-dtype 的 NumPy 和 Pandas 列会保留其现有的 UTC 墙钟时间转换方式。

若要表示不依赖于任一模式的特定时刻，请附加预期的时区，或显式提供纪元整数。

from datetime import datetime, timezone

utc_time = datetime(2023, 6, 15, 10, 30, tzinfo=timezone.utc)
client.insert("events", [[utc_time]], column_names=["event_time"])

naive_time = datetime(2023, 6, 15, 10, 30)
epoch_timestamp = int(naive_time.replace(tzinfo=timezone.utc).timestamp())
client.insert("events", [[epoch_timestamp]], column_names=["event_time"])

不含时区信息的 datetime 查询参数使用独立的 naive_datetime_binding 设置。默认的 "wall" 模式会按原样发送日期时间字段，不进行主机本地时区转换。请参阅参数 argument部分。

带有时区元数据的 DateTime 列

ClickHouse 列可以声明时区元数据，例如 DateTime('America/Denver') 或 DateTime64(3, 'Asia/Tokyo')。这些元数据决定了值在查询时的显示方式。

插入带时区信息的值时，ClickHouse Connect 会保留其所表示的时刻。对于不带时区信息的值，naive_datetime_insert 设置决定使用进程时区还是列时区。查询时，结果会使用该列的时区，除非通过 column_tzs 参数为特定列提供覆盖设置。query_tz 参数不会覆盖列中已声明的时区。

from datetime import datetime
from zoneinfo import ZoneInfo

client.command(
    "CREATE TABLE events_with_timezone "
    "(event_time DateTime('America/Los_Angeles')) "
    "ENGINE Memory"
)

data = datetime(2023, 6, 15, 10, 30, tzinfo=ZoneInfo("America/New_York"))
client.insert("events_with_timezone", [[data]], column_names=["event_time"])

result = client.query("SELECT event_time FROM events_with_timezone")
returned = result.first_row[0]
assert returned.hour == 7
assert returned.tzinfo == ZoneInfo("America/Los_Angeles")

File 插入

clickhouse_connect.driver.tools.insert_file 会以流式方式将本地文件插入现有表中，并由 ClickHouse 负责解析。

参数	类型	默认值	说明
`client`	`Client`	必填	用于执行插入的同步客户端。
`table`	str	必填	简单表名或带数据库限定的目标表。
`file_path`	str	必填	输入文件的本地路径。
`fmt`	str	`"CSV"` or `"CSVWithNames"`	输入格式。提供 `column_names` 时，默认值为 `"CSV"`；否则默认值为 `"CSVWithNames"`。
`column_names`	Sequence[str]	`None`	文件中表示的列。对于包含列名的格式，不需要提供。
`database`	str	`None`	当表名未带数据库限定时使用的目标数据库。
`settings`	dict	`None`	参见 Settings 参数。
`compression`	str	`None`	文件当前使用的压缩格式，例如 `"zstd"`、`"lz4"` 或 `"gzip"`。对于 `.gz` 和 `.gzip` 文件名，会自动推断为 gzip。

可通过 settings 传递输入格式设置，例如 input_format_allow_errors_ratio 和 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": 0.2,
        "input_format_allow_errors_num": 5,
    },
)

对于 AsyncClient，使用相同参数 await insert_file_async：

from clickhouse_connect.driver.tools import insert_file_async

await insert_file_async(async_client, "example_table", "my_data.csv")

async 辅助函数会先在工作线程中读取文件，再等待 raw_insert，因此文件内容会保存在内存中。