Python Generators 與 Iterators：處理大型資料而不讓記憶體爆掉
Python Generators 與 Iterators：處理大型資料而不讓記憶體爆掉
當 Python 腳本開始消耗數百 MB 的 RAM 時，直覺反應通常是改用更快的語言或更進階的資料庫。但大多數情況下，真正的問題簡單得多：程式碼一次就把整個資料集載入記憶體。Generators（Python 的惰性求值主力）讓你一次只處理一筆資料，它們是你能加入 Python 工具箱中最高槓桿的概念之一。
顯而易見的記憶體問題
考慮一個常見任務：讀取大型日誌檔案並計算有多少行包含「error」這個字。最直接的寫法看起來無害：
def count_errors(path):
with open(path) as f:
lines = f.readlines() # loads EVERYTHING into memory
return sum(1 for line in lines if "error" in line.lower())
Enter fullscreen mode
Exit fullscreen mode
對 10 MB 的檔案來說這執行得很好。但對 4 GB 的日誌檔案，readlines() 會很開心地試圖把全部 4 GB 都放在 RAM 中——而在記憶體上限只有 2 GB 的共享伺服器上，這個行程就會被終止。修正方式只要改一個字：
def count_errors(path):
with open(path) as f:
return sum(1 for line in f if "error" in line.lower())
Enter fullscreen mode
Exit fullscreen mode
直接對檔案物件進行疊代會一次產生一行。作業系統會串流它，而你的記憶體用量不管檔案多大都會保持平穩。這就是 generator 的本質：它計算並產生一個值，然後暫停，直到下一個值被請求為止。
Generator 到底是什麼？
Generator 是一種特殊的 iterator，可以透過 generator 函式或 generator 運算式建立。其定義特徵是 yield 關鍵字。當函式包含 yield 時，呼叫它並不會執行函式主體——而是回傳一個你可以疊代的 generator 物件。
def read_large_file(path):
"""Yield one line at a time from a potentially huge file."""
with open(path) as f:
for line in f:
yield line.strip()
Enter fullscreen mode
Exit fullscreen mode
與會建立並回傳完整 list 的普通函式比較：
def read_all_lines(path):
with open(path) as f:
return [line.strip() for line in f] # materializes the whole list
Enter fullscreen mode
Exit fullscreen mode
Generator 版本幾乎使用固定記憶體。List 版本則會隨著檔案大小而擴展。對 5 GB CSV 進行互動式探索時，這種差異決定了工具是能即時回應還是讓機器凍結。
Generator 運算式與 List Comprehensions
Python 提供簡潔的語法來建立 generator，其外觀與 list comprehensions 相似。唯一的差別是使用小括號而非中括號：
squares_list = [x*x for x in range(1_000_000)] # list: ~8 MB allocated at once
squares_gen = (x*x for x in range(1_000_000)) # generator: lazy, one at a time
Enter fullscreen mode
Exit fullscreen mode
要注意一個細微之處：generator 是單次使用的。一旦被消耗，就會耗盡。如果你需要疊代兩次，就必須重新建立 generator 或把結果存成 list。
gen = (x for x in range(5))
print(list(gen)) # [0, 1, 2, 3, 4]
print(list(gen)) # [] -- already exhausted!
Enter fullscreen mode
Exit fullscreen mode
真正能節省記憶體的實用模式
1. 使用 islice 進行分塊處理
有時候你確實需要 generator 的一小段，但切片語法只能用在 sequence 上。itertools.islice 函式會惰性地逐步處理 generator，並回傳固定數量的項目：
from itertools import islice
def process_in_chunks(collection, chunk_size=1000):
iterator = iter(collection)
while True:
chunk = list(islice(iterator, chunk_size))
if not chunk:
break
process_chunk(chunk) # your batch logic here
Enter fullscreen mode
Exit fullscreen mode
這個模式非常適合將記錄分批送進資料庫，使用可管理的交易，而不是一次提交巨量資料。
2. 串流聚合
因為 generator 是惰性產生值的，你可以把多個轉換串接起來，而不會具體化任何中間 list：
import re
from collections import Counter
def log_errors(path):
with open(path) as f:
pattern = re.compile(r"errors*:s*(w+)")
for line in f:
match = pattern.search(line)
if match:
yield match.group(1)
code_counts = Counter(log_errors("app.log"))
print(code_counts.most_common(5))
Enter fullscreen mode
Exit fullscreen mode
Counter 只需要與不同錯誤代碼的數量成比例的記憶體，這通常很小，而不是與日誌行數成比例。
3. yield from 快捷方式
Generator 委託能讓一個 generator 乾淨地交接給另一個。yield from 會把來自內部 iterable 的每個項目轉發出去，這在組合資料管線時特別有用：
def read_lines(paths):
for path in paths:
with open(path) as f:
yield from f # delegate to the inner iterable
for line in read_lines(["a.log", "b.log", "c.log"]):
...
Enter fullscreen mode
Exit fullscreen mode
什麼時候不該使用 Generator
Generator 不是萬靈丹。了解它們的限制可以避免誤用：
| 情境 | 該使用 generator 嗎？ | 原因 |
|---|---|---|
| 大型檔案、即時串流、無限序列 | ✅ 是 | 固定記憶體勝出 |
| 需要對元素進行隨機存取 | ❌ 否 | Generator 沒有索引 |
| 需要對相同資料疊代兩次 | ⚠️ 有時 | 必須重新建立或儲存 |
| 小型資料集 | 🤷 皆可 | 額外負擔不值得 |
| 隨機存取 / 回溯 | ❌ 否 | 只能單次通過 |
Generator 是一種單向串流。你無法倒帶，無法在不經過 0 到 4 的元素的情況下跳到第 5 個元素，也無法在不耗盡它的情況下知道它的長度。如果你的演算法需要隨機存取，請保留 list 或使用其他結構。
完整實作範例：在固定記憶體下進行日誌分析
讓我們把這些片段組合起來。這個腳本讀取大型應用程式日誌、計算錯誤等級，並回報前五名的錯誤類型——同時不管檔案大小如何，都能保持記憶體用量平穩：
import re
from collections import Counter
from itertools import islice
PATTERN = re.compile(r"[(?P<level>w+)]s+.*?b(?P<code>w+Error)b", re.IGNORECASE)
def scan_errors(path):
with open(path) as f:
for line in f:
match = PATTERN.search(line)
if match:
yield match.groupdict()
def report(path, limit=5):
level_counts = Counter()
code_counts = Counter()
for chunk in iter(lambda: list(islice(scan_errors(path), 5000)), []):
for entry in chunk:
level_counts[entry["level"]] += 1
code_counts[entry["code"]] += 1
print("Levels:", level_counts.most_common())
print("Top codes:", code_counts.most_common(limit))
if __name__ == "__main__":
report("app.log")
Enter fullscreen mode
Exit fullscreen mode
iter(lambda: list(islice(...)), []) 迴圈是一種簡潔的方式，能拉取固定大小的批次直到 generator 耗盡。每個批次都很小，處理完後會被釋放，然後才處理下一個批次。
進階提示：將值傳送進 Generator
Generator 也可以透過 send() 接收 值，這讓它們變成輕量級的 coroutine。這在簡單的資料處理中很少需要，但它解鎖了雙向通訊：
def accumulator():
total = 0
while True:
received = yield total # yield current total, then wait for input
if received is not None:
total += received
acc = accumulator()
print(next(acc)) # 0 -- prime the generator
print(acc.send(10)) # 10
print(acc.send(5)) # 15
Enter fullscreen mode
Exit fullscreen mode
這個模式是更進階非同步模式和有狀態處理管線的基礎。對大多數日常任務來說你永遠不會需要它，但知道它的存在能幫助你在函式庫程式碼中遇到時認出底層機制。
使用類別建立自己的 Iterator
如果你需要自訂的疊代行為並結合其他方法，可以建立一個實作 __iter__ 和 __next__ 的 iterator 類別。Generator 函式通常比較簡單，但類別形式在需要更豐富狀態的情況下值得了解：
class Countdown:
def __init__(self, start):
self.current = start
def __iter__(self):
return self
def __next__(self):
if self.current <= 0:
raise StopIteration
self.current -= 1
return self.current
Enter fullscreen mode
Exit fullscreen mode
惰性資料處理的實用檢查清單
當你接手一個被記憶體淹沒的腳本時，請執行以下檢查清單：
- 將
readlines()替換為直接對檔案物件進行疊代。 - 將餵給
sum、any、all或Counter的 list comprehensions 轉換為 generator 運算式。 - 將資料庫操作分批進行，而不是逐行插入或一次全部插入。
- 使用
sys.getsizeof在樣本上驗證——但請注意 generator 不會回報它們「可能」的內容，所以請測量你正在替換的 list 版本。 - 如果有可用的記憶體工具就用它來分析；否則就用系統監視器觀察常駐集大小。
總結
Generator 讓 Python 能夠處理否則會耗盡記憶體的資料集，並鼓勵乾淨、串流的程式設計風格。核心概念雖然小卻很強大：
-
yield將函式轉變成惰性 generator。 - Generator 運算式與 comprehensions 相似，但不會積極地建立任何東西。
- 單次通過、串流演算法與 generator 完美契合。
- 使用
itertools.islice進行惰性分塊，並使用yield from進行委託。 - 當你需要隨機存取或重複疊代時，請避免使用 generator。
下次當腳本變得極慢或因記憶體不足而死亡時，找出那個隱藏的 readlines()。用 generator 替換它通常只要改兩行，就能將脆弱的腳本轉變成能擴展到任意大小檔案的版本。