使用Multiprocessing.Pool执行ClickHouse查询时遇614错误求助

代码里的明显问题 #

• 导入错误：from cimport Pool 是写错了，应该是 from multiprocessing import Pool；另外代码里用了os.cpu_count()却没导入os库，会先触发NameError。
• 语法错误：args = [(arg_11, arg_21)]. 末尾多了个句号，属于语法错误，会直接导致代码跑不起来。
• 连接对象复用问题：clickhouse_driver的Client对象不是进程安全的，你在主进程创建了client，然后让所有子进程共享这个对象，会导致连接状态混乱，这是触发614错误的核心原因。

ClickHouse错误码614的含义 #

这个错误码对应UNEXPECTED_PACKET_FROM_CLIENT，说白了就是服务器收到了客户端发过来的乱七八糟的数据包，大概率是因为多进程共用同一个连接对象，导致发送的数据包格式被打乱了。

解决办法 #

1. 先修正代码基础错误 #

把导入、语法问题先改好，同时给每个子进程单独创建Client对象：

from multiprocessing import Pool
import os
from clickhouse_driver import Client

# 这里补充你的连接参数
conn_dct = {"host": "你的ClickHouse地址", "user": "用户名", "password": "密码"}

def foo(arg_1, arg_2):
    # 每个子进程自己创建独立的Client实例
    client = Client(**conn_dct)
    # 注意VALUES后面要加括号，符合SQL语法
    client.insert(f'INSERT INTO TABLE VALUES ({arg_1}, {arg_2})')
    # 显式断开连接，避免资源泄露
    client.disconnect()

# 去掉末尾的句号
args = [(arg_11, arg_21)]
pool = Pool(os.cpu_count()-1)
pool.starmap(foo, args)
pool.close()
pool.join()

2. 优化插入逻辑（可选） #

如果是批量插入场景，别单条插，把一批数据打包成一次插入，能大幅减少连接开销，比如：

def foo(batch_data):
    client = Client(**conn_dct)
    # 把一批数据拼接成合法的SQL值列表
    values_str = ', '.join([f'({a}, {b})' for a, b in batch_data])
    client.insert(f'INSERT INTO TABLE VALUES {values_str}')
    client.disconnect()

# 把数据分成批次传入
batch_args = [[(arg_11, arg_21), (arg_12, arg_22)]]
pool.starmap(foo, batch_args)

是否需要更换并行库？ #

完全不需要，multiprocessing.Pool完全够用，问题根本不在并行库上，而是你对ClickHouse连接对象的使用方式错了，加上代码本身有语法/导入错误。

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