Python用for循环按年分批插入ClickHouse数据出现超时错误如何解决

你遇到的报错核心是同步客户端执行长耗时查询时，内置的ping保活机制检测超时主动断开连接，不需要第一时间切换异步方案，优先优化现有逻辑即可解决问题。

Error on %Server% ping: timed out. Connection was closed, reconnecting.

首先现有代码存在明显缺陷：你用pd.date_range生成了1979到2029年的天级日期序列，循环中每次取对应日期的年份执行插入，会导致同一个年份的插入语句重复执行数百次，大量无效请求不仅浪费资源，也大幅提升了超时概率。

方案1：优化同步实现（推荐优先使用，改造成本最低） #

修复重复执行逻辑，同时调整客户端超时参数即可解决绝大多数场景的超时问题，对应代码如下：

import pandas as pd
from clickhouse_driver import Client

# 初始化客户端时调整超时参数
client = Client(
    host=host, 
    user=user, 
    password=password, 
    database='default',
    send_receive_timeout=3600,  # 单次请求最长等待时间，单位秒，可按需调整
    ping_timeout=3600  # ping超时时间，和上述参数保持一致即可
)

# 直接生成唯一年份列表，不需要生成天级序列
years = range(1979, 2030)
for year in years:
    query = f"insert into t2 select * from t1 where y = {year}"
    client.execute(query)

client.disconnect()

方案2：异步实现方案（适合需要并发执行多个批次的场景） #

如果单批次插入耗时过长，想要同时跑多个批次提升效率，可以使用异步客户端实现，注意要控制并发数避免压垮服务端，对应代码如下：

import asyncio
from aiochclient import ChClient
from aiohttp import ClientSession

async def insert_by_year(year):
    async with ClientSession() as session:
        async with ChClient(
            session,
            url=f"http://{host}:8123",
            user=user,
            password=password,
            database="default",
        ) as client:
            await client.execute(f"insert into t2 select * from t1 where y = {year}")

async def main():
    years = range(1979, 2030)
    # 控制并发数，这里设置同时跑3个批次，可根据服务端性能调整
    semaphore = asyncio.Semaphore(3)
    async def bounded_insert(year):
        async with semaphore:
            await insert_by_year(year)
    tasks = [bounded_insert(year) for year in years]
    await asyncio.gather(*tasks)

if __name__ == "__main__":
    asyncio.run(main())

额外优化建议 #

• 提前给源表t1的年份字段y建立索引，大幅加快单批次查询过滤速度
• 如果单年份数据量超过1亿，可以进一步拆分批次到季度或者月度，降低单次请求的执行耗时
• 执行前可以先单独测试一个年份的插入耗时，再对应调整超时参数

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