如何用Python将千万级SQL插入语句拆分为指定大小批次

核心逻辑 #

把包含大量数据的可迭代对象按指定chunk_size拆分，每个批次生成一条多值插入的SQL语句，方便后续用线程池等方式并行执行。

代码实现 #

1. 可迭代对象拆分函数

先实现通用的拆分函数，将大的可迭代数据切成指定大小的批次：

def chunk_iterable(iterable, chunk_size):
    chunk = []
    for item in iterable:
        chunk.append(item)
        if len(chunk) == chunk_size:
            yield chunk
            chunk = []
    # 处理最后一批不足chunk_size的数据
    if chunk:
        yield chunk

2. 批量生成插入语句

基于拆分函数，生成符合要求的SQL插入语句：

def generate_batch_insert_queries(table_name, column_names, data_iterable, chunk_size):
    # 拼接基础SQL前缀
    base_query = f"insert into {table_name} ({', '.join(column_names)}) values "
    # 遍历每个数据批次
    for chunk in chunk_iterable(data_iterable, chunk_size):
        # 将批次内的每个数据转为(值)格式，再用逗号连接
        values_clause = ', '.join([f"({item})" for item in chunk])
        yield base_query + values_clause

3. 使用示例

针对你提供的1000万条整数数据场景：

# 示例源数据（1000万条整数）
data_list = [i for i in range(0, 10000000)]
# 每个批次的大小，可根据数据库配置调整
chunk_size = 1000

# 生成所有批次的插入语句
batch_queries = generate_batch_insert_queries('test_db', ['test_col'], data_list, chunk_size)

# 遍历执行（此处可替换为线程池并行执行逻辑）
for query in batch_queries:
    # 示例：打印语句，实际中替换为数据库执行代码
    # cursor.execute(query)
    print(query)

注意事项 #

• 批次大小调整：不同数据库对单条SQL语句的长度有上限（比如MySQL的max_allowed_packet配置），建议根据数据库实际情况调整chunk_size，一般1000-10000条是比较稳妥的范围。
• SQL注入防护：如果数据包含不可信的字符串内容，不要直接拼接SQL语句，改用参数化查询更安全。以MySQL为例，参数化版本的生成逻辑如下：

def generate_safe_batch_queries(table_name, column_names, data_iterable, chunk_size):
    base_query = f"insert into {table_name} ({', '.join(column_names)}) values "
    # 生成单个值的占位符，单列就是(%s)
    single_placeholder = ', '.join(['%s' for _ in column_names])
    for chunk in chunk_iterable(data_iterable, chunk_size):
        # 生成批次对应的占位符串
        batch_placeholder = ', '.join([f"({single_placeholder})" for _ in chunk])
        # 整理参数列表（单列数据直接转元组即可）
        params = tuple(chunk)
        yield base_query + batch_placeholder, params

# 使用时通过参数化执行
# for query, params in generate_safe_batch_queries(...):
#     cursor.execute(query, params)

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