Python 3.12 起,
itertools.batched(iterable, n)就是把資料切成固定大小批次的標準做法,每一批回傳 tuple,最後一批不滿 n 個也照樣送出;3.13 又加了strict=True,最後一批不滿就丟ValueError。要跑在 3.12 以前的環境,下面islice那一版可以接任何 iterable,其餘幾種則要求輸入能問長度、能切片。
把一串資料切成固定大小的小份,需求到處都是:一次送一百筆進 API、一次寫一千列進資料庫、一次讀一批進模型。Python 3.12 以前,標準函式庫沒有現成的東西,這件事一直是各寫各的。
itertools.batched
Python 3.12 把 batched 加進 itertools,簽名是 batched(iterable, n, *, strict=False)。文件寫:「Batch data from the iterable into tuples of length n. The last batch may be shorter than n.」
from itertools import batched
print(list(batched(range(15), 4)))
## [(0, 1, 2, 3), (4, 5, 6, 7), (8, 9, 10, 11), (12, 13, 14)]
15 筆切成每批 4 個,前三批是滿的,最後剩 3 個就給 3 個。每一批的型別是 tuple,拿到手之後不能直接 append 進去,要改內容得先 list(batch) 轉一次;想省這一步,more-itertools 的 chunked 直接回傳長度 n 的 list,代價是多一個要裝的套件。
最後一批不滿的情形,有些場合不能就這樣送出去。3.13 之後可以把 strict 打開:
list(batched(range(15), 4, strict=True))
## ValueError: batched(): incomplete batch
最後一批比 n 短的時候,strict=True 就丟 ValueError。對面的介面要求每次剛好 n 筆的時候,交給它擋,比自己在迴圈裡數長度乾淨。這個參數目前是 3.13 才有的,3.12 只有前面那兩個。
yield
3.12 以前,這件事都得自己寫。最直覺的一種是拿 range 的第三個參數當步長,每次切一段出來 yield:
def chunks1(input_list, n):
for i in range(0, len(input_list), n):
yield input_list[i:i + n]
input_list = [i for i in range(0, 15)]
print(list(chunks1(input_list, 4)))
## [[0, 1, 2, 3], [4, 5, 6, 7], [8, 9, 10, 11], [12, 13, 14]]
range(0, len(input_list), n) 產生 0、4、8、12 這幾個起點,每個起點往後切 n 個,切到尾端不夠就切多少算多少。這個寫法本身是 generator,呼叫的當下不會算,外面包 list() 才會一次跑完。它對輸入的要求是能問長度、能切片,list、tuple、字串都符合。
一行for迴圈
同一段邏輯寫成列表推導式,就縮成一行,差別在於它一次把所有小份都算好放進一個 list:
input_list = [i for i in range(0, 15)]
n = 3
output_list = [input_list[i:i+ n] for i in range(0, len(input_list), n)]
print(output_list)
## [[0, 1, 2], [3, 4, 5], [6, 7, 8], [9, 10, 11], [12, 13, 14]]
n 換成 3,切出來就是五份。資料量大的時候整份結果都在記憶體裡,上面那個 generator 版本則是要一份才給一份。
iterable
要接任何 iterable,就得換一條路:把輸入轉成 iterator,每次用 islice 往前拿 n 個包成 tuple。
from itertools import islice
def chunks2(input_iter, n):
input_list = iter(input_iter)
return iter(lambda: tuple(islice(input_list, n)), ())
input_list = [i for i in range(0, 15)]
n = 4
print(list(chunks2(input_list, n)))
## [(0, 1, 2, 3), (4, 5, 6, 7), (8, 9, 10, 11), (12, 13, 14)]
iter() 除了常見的單參數用法,還有一個 iter(callable, sentinel) 的兩參數形式,會反覆呼叫前面那個函式,直到回傳值等於 sentinel 為止。這裡的 sentinel 是空 tuple:輸入拿光之後,islice 就再也拿不到東西,tuple() 出來剛好是空的,迴圈停在那裡。
Numpy
NumPy 也有現成的 array_split,不過它問的問題和前面幾種不太一樣:
import numpy as np
input_list = [i for i in range(0, 15)]
np.array_split(input_list, 5)
## [array([0, 1, 2]),
## array([3, 4, 5]),
## array([6, 7, 8]),
## array([ 9, 10, 11]),
## array([12, 13, 14])]
前面幾種寫法的 n 都是「每一批幾個」,array_split 的第二個參數是「切成幾份」。15 筆切成 5 份剛好每份 3 個,看起來和每批 3 個是同一回事,換成不整除的數字就分開了。長度 l 的陣列要切成 n 份,出來會有 l % n 份是 l//n + 1 個,其餘每份 l//n 個。15 筆切成 4 份會是 4、4、4、3,餘數攤在前面幾份,batched 則是把不滿的那批留在最後。回傳的每一份也是 ndarray,後面要當一般序列用得自己轉。
沒有長度的輸入
把輸入換成 generator,列表推導式那一版會在 len() 就停住:
def gen():
yield from range(15)
g = gen()
[g[i:i + 4] for i in range(0, len(g), 4)]
## TypeError: object of type 'generator' has no len()
generator 只能一個一個往前拿,拿過的就過去了,既不能回頭數還剩幾個,也不能用索引切片。同樣性質的東西在日常程式裡不少:開著的檔案物件、csv.reader、資料庫 cursor、一個還在收的網路回應,這一類物件的共同點可以參考 Python 的 iterable。這種輸入正好是 islice 那一版和 batched 的守備範圍:
# chunks2 沿用上面那段的定義
print(list(batched(gen(), 4)))
print(list(chunks2(gen(), 4)))
## [(0, 1, 2, 3), (4, 5, 6, 7), (8, 9, 10, 11), (12, 13, 14)]
## [(0, 1, 2, 3), (4, 5, 6, 7), (8, 9, 10, 11), (12, 13, 14)]
兩邊輸出一樣,底下的機制也一樣。文件也給了一份 batched 的等價實作,核心就是同一句 tuple(islice(iterator, n)),只是停止條件寫成 while 迴圈,chunks2 則交給 iter() 的 sentinel。
還有一個從輸出看不出來的性質:讀取是邊拿邊讀的,只讀到剛好填滿一批為止。拿一個會邊跑邊印字的 generator 來看比較清楚:
def counting():
for i in range(100):
print("read", i)
yield i
it = batched(counting(), 3)
print(next(it))
print(next(it))
## read 0
## read 1
## read 2
## (0, 1, 2)
## read 3
## read 4
## read 5
## (3, 4, 5)
來源有一百筆,拿第一批的時候只讀了三筆,拿第二批再往前讀三筆。來源換成檔案、網路請求、或一個還在跑的查詢時,這個差別會直接反映在記憶體用量和第一批出現的時間上:整份讀完再切,第一批要等全部到齊;邊讀邊切的話,湊滿一批就先送出去了。
選哪一種
目前手上的環境是 3.12 以上,就用 batched,自己寫的那幾種留給還沒升上去的地方。3.12 以前,islice 那一版照抄就行,它吃的輸入範圍和 batched 一樣,只是要自己多寫一個函式。輸入固定是 list、又只是要分頁顯示,列表推導式那一行最短。至於要的是「切成 k 份」而不是「每份 n 個」,那是 array_split 的題目。
輸入是一個大到不想整份讀進來的檔案,分批這件事通常在讀取那一層就處理掉,不必先讀成 list 再切。pandas.read_csv() 的 chunksize 就是這樣用的:給了它一個數字,read_csv() 回傳的是一個可以 for 下去、一次給一塊 DataFrame 的 TextFileReader。
一次送一百筆進 API 的那種情況,3.12 以上寫成 for batch in batched(rows, 100): 就結束了,剩下要記得的是最後一批可能不滿一百。


