使用clickhouse-driver查询内存超限，如何实现查询分块？

当使用clickhouse-driver的query_dataframe()一次性拉取大结果集时，容易触发内存超限错误。以下是几种分块处理的可行方案：

方案一：OFFSET + LIMIT 逐块拉取 #

通过构造带LIMIT和OFFSET的查询，分批获取数据后合并成DataFrame，适合数据量不是极端庞大的场景：

from clickhouse_driver import Client
import pandas as pd

client = Client('your_clickhouse_host')
raw_query = 'CUSTOM SQL QUERY'

# 获取查询结果的列名
columns = [col[0] for col in client.execute(f"DESCRIBE TABLE ({raw_query})")]

chunk_size = 100000  # 可根据内存情况调整
offset = 0
df_list = []

while True:
    chunk_query = f"{raw_query} LIMIT {chunk_size} OFFSET {offset}"
    chunk_data = client.execute(chunk_query)
    if not chunk_data:
        break
    df_list.append(pd.DataFrame(chunk_data, columns=columns))
    offset += chunk_size

final_df = pd.concat(df_list, ignore_index=True)

注意：OFFSET在处理超大规模数据时效率较低，因为ClickHouse需要跳过前序行。

方案二：按分区/范围键分块（推荐） #

利用表的分区键（如日期）或唯一范围键（如ID）分块查询，效率远高于OFFSET方案：

按分区键查询（以日期分区为例） #

client = Client('your_clickhouse_host')
raw_query = 'CUSTOM SQL QUERY'

# 获取所有分区值
partitions = client.execute("SELECT DISTINCT date FROM your_target_table")
df_list = []

for (date_val,) in partitions:
    chunk_query = f"{raw_query} WHERE date = '{date_val}'"
    df_list.append(client.query_dataframe(chunk_query))

final_df = pd.concat(df_list, ignore_index=True)

按范围键查询（以ID为例） #

client = Client('your_clickhouse_host')
raw_query = 'CUSTOM SQL QUERY'

# 获取ID的最小/最大值
min_id, max_id = client.execute("SELECT MIN(id), MAX(id) FROM (SELECT id FROM your_target_table)")[0]

step = 100000  # 每块覆盖的ID范围
df_list = []
current_min = min_id

while current_min <= max_id:
    current_max = current_min + step - 1
    chunk_query = f"{raw_query} WHERE id BETWEEN {current_min} AND {current_max}"
    df_list.append(client.query_dataframe(chunk_query))
    current_min = current_max + 1

final_df = pd.concat(df_list, ignore_index=True)

方案三：使用流式查询接口 #

若你的clickhouse-driver版本支持，可通过query_stream()流式读取数据，逐块生成DataFrame：

from clickhouse_driver import Client
import pandas as pd

client = Client('your_clickhouse_host')
raw_query = 'CUSTOM SQL QUERY'

chunk_size = 100000
current_chunk = []
df_list = []

with client.query_stream(raw_query) as stream:
    columns = stream.columns
    for row in stream:
        current_chunk.append(row)
        if len(current_chunk) >= chunk_size:
            df_list.append(pd.DataFrame(current_chunk, columns=columns))
            current_chunk = []
    # 处理剩余数据
    if current_chunk:
        df_list.append(pd.DataFrame(current_chunk, columns=columns))

final_df = pd.concat(df_list, ignore_index=True)

额外建议 #

• 调整chunk_size：根据本地内存容量灵活设置，避免再次触发内存限制。
• 优化原查询：优先通过WHERE子句过滤不必要的数据，或只选择需要的列，减少返回数据量。

内容的提问来源于stack exchange，提问作者frogseer