itertools.batched() (3.12+) 如何高效分批处理大迭代器

itertools.batched() 更安全，因其不预加载全部数据、内存占用恒定；手写切片易致全量展开，引发 OOM 或阻塞。

batched() 为什么比手写切片更安全

直接用 itertools.batched() 处理大迭代器，核心优势是它不预加载全部数据——它只维持当前批次的元素，内存占用恒定。手写 list(it)[i:i+n] 或 islice 循环容易误触发全量展开，尤其当 it 是生成器或数据库游标时，可能 OOM 或阻塞超时。

常见错误现象：MemoryError、迭代中途卡住、CPU 占用突增但无输出。

使用场景：读取超大 CSV 文件流、分页拉取 API 数据、逐批处理日志行。

batched() 的参数陷阱和替代选择

batched() 只接受两个参数：iterable 和 n（每批元素数），不支持自定义填充、跳过不完整批次等逻辑。遇到末尾不足 n 个元素时，默认返回最后一组（长度 n）。

如果需要补全（比如喂给固定尺寸模型输入），得自己 wrap：

MOKI

MOKI是美图推出的一款AI短片创作工具，旨在通过AI技术自动生成分镜图并转为视频素材。

下载

from itertools import batched, chain, repeat
def batched_padded(iterable, n, fillvalue=None):
for batch in batched(iterable, n):
yield tuple(batch) + tuple(repeat(fillvalue, n - len(batch)))

性能影响：补全逻辑增加少量开销，但远低于转成 list 再 pad；batched() 本身是 C 实现，比纯 Python 循环快 2–3 倍。

和 chunked()（more-itertools）对比要注意什么

more-itertools.chunked() 行为相似，但返回的是 list 而非 tuple，且在 Python

• batched() 返回 tuple，不可变，适合后续 hash 或作为 dict key
• chunked() 返回 list，可原地修改，但多一次内存分配
• batched() 是标准库，无额外依赖；chunked() 需装 more-itertools
• 两者都不缓存整个迭代器，但 chunked() 对某些惰性对象（如 map）兼容性略差

真实大迭代器下的实操建议

处理真正的大迭代器（比如千万级日志行），光靠 batched() 不够，得配合下游消费节奏控制：

• 避免把整批 tuple 一次性塞进 pandas DataFrame —— 改用 pd.concat([pd.DataFrame(batch) for batch in batched(...)], ignore_index=True) 会累积内存；应逐批 .append() 或用 chunksize 参数反向驱动
• 若下游是数据库批量插入，确认驱动是否支持 executemany()；传入 batched() 结果前，先 map(tuple, ...) 转成元组序列（多数驱动要求）
• 调试时别用 print(list(batched(...))) —— 这会强制展开全部批次，失去流式意义
• 注意 n 的取值：太小（如 n=1）导致调用开销占比高；太大（如 n=100000）可能让单批处理时间过长，阻塞协程或超时；推荐从 1000 起测，按吞吐和延迟调优

最易被忽略的一点：batched() 不感知迭代器内部状态，如果源迭代器本身有副作用（如文件指针移动、网络重连），批次边界可能打乱业务语义——这时必须用带状态封装的自定义迭代器，而不是依赖 batched() 切分。