如何用Generator读取多行？超大规模CSV分批处理需求

嘿，我太懂你处理超大CSV时的崩溃感了——1100万行直接塞内存绝对行不通，用生成器分批读本来是完美解法，结果踩了“只能迭代一次”的坑，还因为对生成器不熟折腾了半天对吧？我来给你把这个问题彻底捋顺。

核心问题：为什么你的临时生成器只能跑一次？ #

生成器本质是一次性迭代器，一旦你把它的元素遍历完，这个生成器对象就耗尽了，没法再从头开始。解决办法很简单：把生成器逻辑封装成函数，每次调用函数都会创建一个全新的生成器实例，自然就能多次迭代啦！

针对你的需求的完整实现 #

考虑到你用numpy，而且每行固定29个值，我给你写个适配的生成器函数，既能自定义批次大小，还能重复调用：

import csv
import numpy as np

def batch_csv_reader(file_path, batch_size=50):
    # 每次调用函数都会重新打开文件，生成新的生成器
    with open(file_path, 'r', newline='', encoding='utf-8') as f:
        csv_reader = csv.reader(f)
        # 可选：如果你的CSV有表头，先读出来返回（不需要就删掉这两行）
        header = next(csv_reader)
        yield header
        
        batch = []
        for row in csv_reader:
            # 把每行转成numpy数组（根据你的数据类型调整dtype，比如str/int）
            row_np = np.array(row, dtype=np.float64)
            batch.append(row_np)
            
            # 攒够批次大小就返回
            if len(batch) == batch_size:
                yield np.vstack(batch)  # 把列表转成二维numpy数组
                batch = []
        
        # 处理最后一批不足batch_size的剩余行
        if batch:
            yield np.vstack(batch)

怎么用这个生成器？ #

调用方式超简单，而且想迭代多少次就多少次，批次大小随便改：

# 第一次迭代：批次设为50
print("第一次处理，批次50：")
for batch in batch_csv_reader('你的超大文件.csv', batch_size=50):
    # 这里写你的处理逻辑，比如数据清洗、模型训练等
    print(f"当前批次行数：{batch.shape[0]}，列数：{batch.shape[1]}")

# 第二次迭代：批次改成2
print("\n第二次处理，批次2：")
for small_batch in batch_csv_reader('你的超大文件.csv', batch_size=2):
    print(f"当前小批次行数：{small_batch.shape[0]}")

额外小贴士 #

• 如果不需要numpy，想要纯列表格式，把里面的numpy转换代码删掉就行，直接返回列表批次。
• 注意CSV的编码和分隔符，如果你的文件用的是制表符分隔，就把csv.reader改成csv.reader(f, delimiter='\t')。
• 生成器的调用除了for循环，也可以用next()手动取批次，但for循环会自动处理迭代结束的StopIteration异常，更省心。

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