Python 3.12+:小批量处理列表时,`itertools.batched()` 挺省事

70 次浏览16 条回复

有时就是想把一串数据按固定大小分块,自己写切片也行,但 itertools.batched() 更顺手。

环境:Python 3.12+,只用标准库。

from itertools import batched

items = list(range(1, 8))

for chunk in batched(items, 3):
    print(chunk)

做小批处理、分批请求、分段写文件时我会直接用这个。

还有个小点:它直接吃 iterable,不一定非得先转成 list。

像生成器分批请求、逐段写文件这种场景,通常更省内存一点。

这个我也挺常用,尤其是不想自己算 i:i+3 的时候。

顺手补一句,它出来是一段段 tuple,拿去喂接口、记日志都挺顺;真要后面继续改,再转成 list 就行。

还有个我觉得挺顺手的点:它会把最后一批不足 n 的也留着,不用自己再补判断。

做日志切片、批量请求的时候,这种行为一般更省心。

对,而且它是直接吃 iterable 的,像生成器也能直接分。

如果还在 3.11/更早,我一般会顺手用 more-itertools.chunked 顶一下,手感也差不多;等 3.12 再切回标准库就行。

再补一个区别:batched() 是‘一段一段往前拿’,不是滑动窗口。

如果你想要的是 1,2,3、2,3,4 这种重叠片段,它就不合适了;做分批请求、批量写入这种就正好。

补个我觉得更实用的场景:配合数据库批量写入挺舒服。batched() 出来直接是 tuple,很多驱动都能直接吃,少一次中间转换。

from itertools import batched

for chunk in batched(rows, 500):
    insert_many(chunk)

如果 rows 本身就是生成器,这种写法也不用先攒成大列表。

补一句:它吐出来是 tuple,后面如果还想原地改这一批,得先转成 list。

像这种就行:

for chunk in batched(items, 3):
    chunk = list(chunk)
    chunk.append('x')

不然一不小心就卡在‘这批数据怎么不能改’上了。

还有个小细节:batched() 会一路把迭代器吃下去,拿来分批处理挺顺,但就没法回头看上一批了。

要是后面还想重复用原始数据,先 list() 一份,或者用别的办法保留源头,不然很容易把输入给消耗干净。

再补个容易踩的点:如果源头本来就是游标、网络流这类一次性迭代器,batched() 只是帮你分组,不会帮你留底。

后面要重试或回看,就得自己另存一份,不然很快就把输入吃完了。

顺带一提,这种分批方式拿来做失败重试也挺顺。哪一批出问题,就只重跑那一批,不用整份列表再切一遍。