如何将大型CSV及含50k值的CSV字符串拆分为500条批量？

我来给你写个实用的Python包装脚本，不管你手里是现成的CSV字符串，还是存储在input.csv里的数据，都能按每批500条拆分，调用目标脚本处理，每批完成后自动等待60秒再继续下一批。

核心思路 #

• 先把所有数据加载成列表，避免重复读取文件
• 按每500条为一组拆分批次
• 对每个批次，调用目标脚本处理（支持两种传递方式：命令行参数或标准输入）
• 每批处理完成后，暂停60秒再进行下一批

完整脚本 #

import subprocess
import time
import csv

# 配置项 - 根据你的实际情况修改
BATCH_SIZE = 500
WAIT_SECONDS = 60
TARGET_SCRIPT_PATH = "./your_target_script.py"  # 替换成你的目标脚本路径
# 选择数据来源：要么用CSV字符串，要么读取input.csv
USE_CSV_STRING = False  # 改成True用CSV字符串
CSV_STRING = "23445, 23446, 24567, ..."  # 这里放你的5万条CSV字符串
INPUT_CSV_PATH = "./input.csv"

def load_data():
    """加载所有数据到列表"""
    if USE_CSV_STRING:
        # 处理CSV字符串：去除空格后拆分
        return [item.strip() for item in CSV_STRING.split(",") if item.strip()]
    else:
        # 读取input.csv文件
        data = []
        with open(INPUT_CSV_PATH, "r") as f:
            # 适配两种CSV格式：一行所有值，或者每行一个值
            reader = csv.reader(f)
            for row in reader:
                data.extend([item.strip() for item in row if item.strip()])
        return data

def process_batch(batch):
    """处理单个批次：调用目标脚本"""
    # 把批次转换成逗号分隔的字符串
    batch_str = ",".join(batch)
    
    # 方式1：把批次作为命令行参数传给目标脚本
    # 比如目标脚本接收参数：python target_script.py "123,456..."
    result = subprocess.run(
        [TARGET_SCRIPT_PATH, batch_str],
        capture_output=True,
        text=True
    )
    
    # 方式2：把批次通过标准输入传给目标脚本（如果目标脚本读取stdin）
    # result = subprocess.run(
    #     [TARGET_SCRIPT_PATH],
    #     input=batch_str,
    #     capture_output=True,
    #     text=True
    # )
    
    # 打印处理结果（可选，方便调试）
    print(f"处理批次完成，共{len(batch)}条数据")
    print(f"目标脚本输出：{result.stdout}")
    if result.stderr:
        print(f"目标脚本错误信息：{result.stderr}")

def main():
    data = load_data()
    total_items = len(data)
    print(f"加载完成，共{total_items}条数据")
    
    # 拆分批次
    for i in range(0, total_items, BATCH_SIZE):
        batch = data[i:i+BATCH_SIZE]
        batch_num = (i // BATCH_SIZE) + 1
        print(f"\n开始处理第{batch_num}批次（共{len(batch)}条）")
        
        process_batch(batch)
        
        # 最后一批处理完不需要等待
        if i + BATCH_SIZE < total_items:
            print(f"等待{WAIT_SECONDS}秒后处理下一批...")
            time.sleep(WAIT_SECONDS)
    
    print("\n所有批次处理完成！")

if __name__ == "__main__":
    main()

关键部分说明 #

1. 配置项：你只需要修改顶部的几个变量，比如目标脚本路径、数据来源选择，就能快速适配你的场景
2. 数据加载：load_data()函数会自动处理CSV字符串或文件，过滤掉空值和多余空格，返回干净的数据集
3. 批次处理：process_batch()函数提供了两种给目标脚本传数据的方式，你可以根据目标脚本的输入要求二选一
4. 限流逻辑：每批处理完（除了最后一批），用time.sleep(WAIT_SECONDS)实现60秒的等待，避免请求过于频繁

注意事项 #

• 确保目标脚本有可执行权限，如果是Python脚本，直接用python调用即可
• 脚本兼容两种CSV格式：一行包含所有值，或者每行一个值
• 可以根据需求添加异常处理，比如目标脚本执行失败时重试，或者跳过错误批次
• Windows系统下注意路径写法，用\\分隔或者原始字符串即可

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