26. ジェネレータとイテレータ
テーマ: 大規模ログをメモリに載せずに集計するストリーム処理
学習点: yield, ジェネレータ式, itertools, パイプライン設計, メモリ使用量の比較
依存: 標準ライブラリのみ ／ 難易度: 上級
実行方法
uv run 26_generators_stream.py
スクリプト冒頭の PEP 723 メタデータ（# /// script）により、必要なライブラリは
uv が自動的に仮想環境へ導入します。事前の pip install は不要です。
解説
何をするプログラムか
EC サイトなどの Web サーバが残すアクセスログは、1 日で数百万行・数 GB に達することも珍しくありません。これを丸ごとリストに読み込むとメモリが足りなくなりますが、集計に必要なのは「1 行ずつ読んで捨てる」処理だけです。本スクリプトは 30 万行（約 10.6 MB）の模擬アクセスログを生成し、ジェネレータをつないだパイプラインで、エラー件数の集計・ページ別の応答時間統計・複利計算までを、ファイル全体をメモリに載せずに実行します。
コードの読みどころ
read_lines()はyield from fでファイルを 1 行ずつ返すジェネレータ関数です。呼び出した瞬間には何も読まず、後段が 1 行要求するたびに 1 行だけ読みます。where(parse(read_lines(LOG)), status=500)のように、parse()（辞書化）→where()（絞り込み）→pluck()（列の取り出し）を連結すると、SQL の SELECT-WHERE に似たパイプラインが素の Python で書けます。各段がジェネレータなので、途中結果のリストは一切作られません。- [1] では同じ 100 万要素の 2 乗列を、リスト内包表記
[x ** 2 for x in ...]とジェネレータ式(x ** 2 for x in ...)で作り、sys.getsizeofでオブジェクトサイズを比較します。約 845 万バイト対 200 バイトという差が遅延評価の本質です。 - [3] は
agg.setdefault(r["page"], [0, 0, 0, 10**9])を使い、件数・合計・最大・最小の 4 つを 1 回のループ（1 パス）で同時更新します。 - [4] の
itertools.isliceは「先頭 3 件を取ったら読むのをやめる」ため、巨大ファイルの試し読みに最適です。groupbyが「連続する」同じキーしかまとめない（事前ソートが前提）ことも重要な注意点です。 - [5] の
compound_stream()はwhile True: yield pvの無限ジェネレータですが、next(i for i, v in ... if v >= 200)が条件を満たした時点で評価が止まるため無限ループになりません。
理論的背景
リストは全要素を同時にメモリへ保持するため、メモリ使用量は入力サイズに比例して O(n) で増えます。一方ジェネレータは「現在の状態と次の 1 要素の作り方」だけを保持する遅延評価の仕組みなので、入力が何行あってもメモリ使用量は O(1) のまま一定です。合計・件数・最大値のように 1 パスで求まる統計量なら、この方式でどれほど大きなデータでも処理できます。
実行結果の見方
[1] のオブジェクトサイズの差（8,448,728 バイト対 200 バイト）が、リストとジェネレータの違いを端的に示します。[2] では 30 万行から 500 エラー 8,948 件を抽出し、[5] では年利 3% で元本が 2 倍になるのが 24 年目と出ますが、これは「72 の法則」（72 ÷ 3 = 24 年）と一致します。[6] は同じジェネレータを 2 回 list() すると 2 回目が空になる「使い切り」の性質で、初学者が最もつまずきやすい点です。
ソースコード
# /// script
# requires-python = ">=3.11"
# dependencies = []
# ///
"""26: ジェネレータとイテレータ -------------------------------------------
テーマ: 大規模ログをメモリに載せずに集計するストリーム処理
学習点: yield, ジェネレータ式, itertools, パイプライン設計, メモリ使用量の比較
根拠: リストは全要素をメモリに保持するが、ジェネレータは1要素ずつ
遅延評価するため、メモリ使用量が入力サイズに依存しない（O(1)）。
"""
import itertools
import random
import sys
from collections import Counter
from pathlib import Path
random.seed(42)
OUT = Path("out_26")
OUT.mkdir(exist_ok=True)
LOG = OUT / "access.log"
PAGES = ["/", "/products", "/products/detail", "/cart", "/checkout",
"/login", "/api/search", "/help"]
STATUS = [200] * 88 + [301] * 4 + [404] * 5 + [500] * 3
def make_log(path: Path, n: int = 300_000) -> None:
with path.open("w") as f:
for i in range(n):
uid = random.randrange(1, 8000)
f.write(f'{1735689600 + i * 2} u{uid:05d} '
f'{random.choice(PAGES)} {random.choice(STATUS)} '
f'{random.randint(20, 3000)}\n')
# --- ジェネレータのパイプライン ---
def read_lines(path: Path):
"""1行ずつ読み出す（ファイル全体をメモリに載せない）。"""
with path.open() as f:
yield from f
def parse(lines):
for line in lines:
ts, uid, page, status, ms = line.split()
yield {"ts": int(ts), "uid": uid, "page": page,
"status": int(status), "ms": int(ms)}
def where(records, **cond):
for r in records:
if all(r[k] == v for k, v in cond.items()):
yield r
def pluck(records, key):
return (r[key] for r in records)
def main() -> None:
make_log(LOG)
print(f"ログ生成: {LOG} ({LOG.stat().st_size / 1e6:.1f} MB, "
f"{sum(1 for _ in read_lines(LOG)):,} 行)\n")
print("[1] メモリ効率の比較")
lst = [x ** 2 for x in range(1_000_000)] # リスト内包表記
gen = (x ** 2 for x in range(1_000_000)) # ジェネレータ式
print(f" リスト内包表記のオブジェクト: {sys.getsizeof(lst):>10,} bytes")
print(f" ジェネレータ式のオブジェクト: {sys.getsizeof(gen):>10,} bytes")
print(" ※ 合計値だけ欲しいなら sum(gen) でよく、リスト化は不要。\n")
del lst
print("[2] パイプライン集計 — エラー(5xx)応答の分析")
errs = list(where(parse(read_lines(LOG)), status=500))
print(f" 500エラー: {len(errs):,} 件")
print(f" 発生ページ上位: {Counter(r['page'] for r in errs).most_common(3)}")
print("\n[3] ページ別の応答時間統計（1パスで集計）")
agg = {}
for r in parse(read_lines(LOG)):
s = agg.setdefault(r["page"], [0, 0, 0, 10**9])
s[0] += 1
s[1] += r["ms"]
s[2] = max(s[2], r["ms"])
s[3] = min(s[3], r["ms"])
print(f" {'ページ':<20}{'件数':>9}{'平均ms':>9}{'最大':>8}{'最小':>8}")
for page, (c, tot, mx, mn) in sorted(agg.items(), key=lambda kv: -kv[1][0]):
print(f" {page:<20}{c:>9,}{tot/c:>9.1f}{mx:>8}{mn:>8}")
print("\n[4] itertools の活用")
first10 = list(itertools.islice(parse(read_lines(LOG)), 3))
print(f" islice で先頭3件だけ取得: {[r['page'] for r in first10]}")
# groupby は「連続する」同じキーをまとめる（事前ソートが前提）
seq = "AAABBBCCAAB"
print(f" groupby('{seq}') -> "
f"{[(k, len(list(g))) for k, g in itertools.groupby(seq)]}")
print(f" accumulate 累積和: "
f"{list(itertools.accumulate([10, 20, 30, 40]))}")
print(f" pairwise 隣接ペア: "
f"{list(itertools.pairwise([1, 4, 9, 16]))}")
print(f" chain 連結: {list(itertools.chain([1, 2], [3], [4, 5]))}")
print(f" combinations 組合せ: "
f"{list(itertools.combinations('ABCD', 2))[:4]} ...")
print("\n[5] 無限ジェネレータと遅延評価 — 複利で元本が2倍になる年")
def compound_stream(pv, r):
while True:
yield pv
pv *= 1 + r
doubling = next(i for i, v in enumerate(compound_stream(100, 0.03))
if v >= 200)
print(f" 年利3%: {doubling} 年目に2倍を超える")
print("\n[6] ジェネレータの状態と使い切り")
g = (i for i in range(3))
print(f" 1回目の list(): {list(g)}")
print(f" 2回目の list(): {list(g)} <- 使い切ると空になる")
print(" ※ 再利用したい場合はリスト化するか、ジェネレータ関数を呼び直す。")
if __name__ == "__main__":
main()
実行結果
ログ生成: out_26/access.log (10.6 MB, 300,000 行)
[1] メモリ効率の比較
リスト内包表記のオブジェクト: 8,448,728 bytes
ジェネレータ式のオブジェクト: 200 bytes
※ 合計値だけ欲しいなら sum(gen) でよく、リスト化は不要。
[2] パイプライン集計 — エラー(5xx)応答の分析
500エラー: 8,948 件
発生ページ上位: [('/checkout', 1206), ('/help', 1161), ('/', 1130)]
[3] ページ別の応答時間統計（1パスで集計）
ページ 件数 平均ms 最大 最小
/api/search 37,820 1508.7 3000 20
/ 37,800 1509.5 3000 20
/login 37,652 1512.7 3000 20
/checkout 37,602 1512.6 3000 20
/products 37,548 1519.7 3000 20
/cart 37,222 1506.6 3000 20
/help 37,199 1517.2 3000 20
/products/detail 37,157 1502.1 3000 20
[4] itertools の活用
islice で先頭3件だけ取得: ['/products', '/cart', '/products']
groupby('AAABBBCCAAB') -> [('A', 3), ('B', 3), ('C', 2), ('A', 2), ('B', 1)]
accumulate 累積和: [10, 30, 60, 100]
pairwise 隣接ペア: [(1, 4), (4, 9), (9, 16)]
chain 連結: [1, 2, 3, 4, 5]
combinations 組合せ: [('A', 'B'), ('A', 'C'), ('A', 'D'), ('B', 'C')] ...
[5] 無限ジェネレータと遅延評価 — 複利で元本が2倍になる年
年利3%: 24 年目に2倍を超える
[6] ジェネレータの状態と使い切り
1回目の list(): [0, 1, 2]
2回目の list(): [] <- 使い切ると空になる
※ 再利用したい場合はリスト化するか、ジェネレータ関数を呼び直す。