Pythonのイテレータとジェネレータ：メモリ実測と使い切りの罠

シリーズ：Python 講座 3/6（中級）。 関数とループは書けるけれど、業務バッチの CSV やサーバーログのように、丸ごとメモリに載せるのがためらわれるデータを扱う必要が出てきた人向けです。読み終えると、イテレータを自作し、ジェネレータでパイプラインを組み、メモリの差を勘ではなく数値で示せるようになります。

数百 MB のログを読んでいる途中でスクリプトが落ちた、あるいは同じデータを2回ループしたはずなのに2回目が空だった。この記事はそういう問題を抱えた人のためのものです。どちらも「イテレータプロトコル」という一つの仕組みから来ていて、仕組みが見えれば直すのは簡単です。AtCoder で map(int, input().split()) の結果を二度使って WA になるのも同じ原因です。

先に結論を書きます。1000万個の二乗和で、ジェネレータ式の追跡メモリのピークは 0.3 KiB、同じ処理のリスト内包表記は 390 MiB で、時間差は 5% 以内でした。86 MiB のログをストリーミング処理するとピークは 29.7 KiB、readlines() で処理すると 1,027 MiB で、時間も3倍かかりました。計測はすべて Apple Silicon Mac（M1 Pro）の Python 3.9 で行い、出力はローカルパスだけ ~/demo/ に置き換えています。

イテレータプロトコルを手で書く

for 文はリストのことを知りません。まず iter(obj) を呼んで obj.__iter__() からイテレータを受け取り、そのイテレータに StopIteration が出るまで next()（つまり __next__()）を呼ぶだけです。プロトコルはこれで全部です。

手で書くときは、何度でもループできるイテラブルと、現在位置を覚えているイテレータを分けるのが定石です。

class Countdown:
    """Iterable: each for loop gets a fresh iterator."""
    def __init__(self, start):
        self.start = start
    def __iter__(self):
        return CountdownIter(self.start)

class CountdownIter:
    def __init__(self, n):
        self.n = n
    def __iter__(self):
        return self
    def __next__(self):
        if self.n <= 0:
            raise StopIteration
        self.n -= 1
        return self.n + 1

c = Countdown(3)
print(list(c), list(c))          # reusable
it = iter(c)
print(next(it), next(it), next(it))
try:
    next(it)
except StopIteration:
    print("StopIteration raised")
print(next(it, "default"))

[3, 2, 1] [3, 2, 1]
3 2 1
StopIteration raised
default

イテレータ自身の __iter__ は self を返すので、イテラブルを受け取る場所にイテレータを渡しても動きます。next(it, デフォルト値) は例外の代わりにデフォルト値を返すため、「最初に一致したもの、なければ None」を書くときに便利です。

よくある間違い：コンテナが自分自身をイテレータとして返す

初めて書くときによくあるのが、コンテナに __next__ を付けて __iter__ で self を返す形です。

class Bag:
    def __init__(self, items):
        self.items, self.i = items, 0
    def __iter__(self):
        return self              # BUG: the container is its own iterator
    def __next__(self):
        if self.i >= len(self.items):
            raise StopIteration
        self.i += 1
        return self.items[self.i - 1]

b = Bag(["x", "y"])
print(list(b), list(b))
print([(p, q) for p in b for q in b])

['x', 'y'] []
[]

エラーは出ず、答えだけが間違います。2回目の走査は空になり、同じオブジェクトで二重ループするとカーソルを共有するため結果がまるごと消えます。対処： __iter__ で新しいイテレータを返すか、もっと短く def __iter__(self): yield from self.items と __iter__ をジェネレータにします。

ジェネレータ：プロトコルをタダで手に入れる

関数の中に yield があると、呼び出しの結果はジェネレータオブジェクトになります。本体は最初の next() まで実行されず、yield のたびに値を一つ渡してローカル変数の状態を凍結します。

def countdown(n):
    while n > 0:
        yield n
        n -= 1

g = countdown(3)
print(g)
print(list(g), list(g))

<generator object countdown at 0x1030ce900>
[3, 2, 1] []

クラス20行と同じ結果ですが、最終行の罠も同じです。ジェネレータはイテレータそのものなので、使い切ったら終わりです。

yield from は別のイテラブルに処理を委ねます。再帰的なジェネレータを書くときに自然で、内側のジェネレータの return 値も受け取れます。

def walk(tree):
    for node in tree:
        if isinstance(node, list):
            yield from walk(node)
        else:
            yield node
print(list(walk([1, [2, [3, 4]], 5])))     # [1, 2, 3, 4, 5]

def inner():
    yield 1
    yield 2
    return "inner done"
def outer():
    result = yield from inner()
    print("outer got:", result)
print(list(outer()))   # outer got: inner done → [1, 2]

一行で書くジェネレータ式 (x * x for x in data) も同じものです。唯一の引数なら括弧を省略して sum(x * x for x in data) と書けます。

罠：ジェネレータ内の StopIteration は RuntimeError になる

Python 3.7（PEP 479）以降、ジェネレータ本体の外へ漏れた StopIteration は RuntimeError に変換されます。デフォルト値なしで next() を呼ぶコードが引っかかります。

def first_words(lines):
    it = iter(lines)
    while True:
        line = next(it)          # raises StopIteration at the end...
        yield line.split()[0]

print(list(first_words(["hello world", "good bye"])))

Traceback (most recent call last):
  File "~/demo/pep479.py", line 4, in first_words
    line = next(it)          # raises StopIteration at the end...
StopIteration

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "~/demo/pep479.py", line 7, in <module>
    print(list(first_words(["hello world", "good bye"])))
RuntimeError: generator raised StopIteration

対処は for line in lines: yield line.split()[0] に書き換えるか、next(it, None) が None のときに明示的に return することです。

使い切りイテレータの罠（実際の出力）

これは半日溶かすタイプのバグです。間違えた行では何も起きないからです。「メモリ節約のため」にジェネレータ式でパースし、2回ループすると：

rows = ["alice,30", "bob,25", "carol,41"]
parsed = (line.split(",") for line in rows)
names = [p[0] for p in parsed]
ages  = [int(p[1]) for p in parsed]
print(names, ages)
print(sum(ages) / len(ages))

['alice', 'bob', 'carol'] []
Traceback (most recent call last):
  File "~/demo/trap.py", line 6, in <module>
    print(sum(ages) / len(ages))
ZeroDivisionError: division by zero

エラーは本当の間違いの2行下で、ループとは無関係に見える割り算のエラーとして現れます。Python 3 の map、filter、zip、ファイルオブジェクト、csv.reader もすべてイテレータなので同じことが起きます。競技プログラミングでおなじみの形で再現すると：

nums = map(int, ["3", "1", "2"])
print(max(nums))
print(min(nums))

3
Traceback (most recent call last):
  File "~/demo/trap2.py", line 3, in <module>
    print(min(nums))
ValueError: min() arg is an empty sequence

似た罠がさらに二つあります。3 in iter([1, 2, 3, 4]) は True を返しつつ 1, 2, 3 を消費するので、続く list() は [4] になります。len(zip(...)) は TypeError: object of type 'zip' has no len() です。イテレータは自分を消費しないと長さが分かりません。

対処は状況で選びます。

itertools の必須関数

itertools はイテレータを受け取りイテレータを返す部品集なので、中間リストを作らずに組み合わせられます。覚えておきたいものの Python 3.9 での実際の出力です。

import itertools as it

print(list(it.islice(it.count(10, 5), 4)))
print(list(it.chain([1, 2], (3, 4), "ab")))
rows = [("ja", "tokyo"), ("ja", "osaka"), ("ko", "seoul"), ("ja", "nagoya")]
print([(k, [c for _, c in g]) for k, g in it.groupby(rows, key=lambda r: r[0])])
print([(k, [c for _, c in g]) for k, g in it.groupby(sorted(rows), key=lambda r: r[0])])
print(list(it.accumulate([3, 1, 4, 1, 5])))
print(list(it.zip_longest("abc", [1], fillvalue="-")))

[10, 15, 20, 25]
[1, 2, 3, 4, 'a', 'b']
[('ja', ['tokyo', 'osaka']), ('ko', ['seoul']), ('ja', ['nagoya'])]
[('ja', ['nagoya', 'osaka', 'tokyo']), ('ko', ['seoul'])]
[3, 4, 8, 9, 14]
[('a', 1), ('b', '-'), ('c', '-')]

groupby の2行を見比べてください。groupby は連続した同じキーしかまとめないので、ソートしないと ja が2回出ます。SQL の GROUP BY やExcel のピボットのつもりで使うと間違えます。無限ストリームでも動くための設計なので、先にソートするか、メモリに収まるなら dict/Counter を使います。

count は無限なので islice や takewhile で切る必要があります。DB へ1000件ずつ INSERT するときなどに使うチャンク関数は、どんなイテレータでも動きます。

def chunks(iterable, size):
    iterator = iter(iterable)
    while True:
        block = list(it.islice(iterator, size))
        if not block:
            return
        yield block
print(list(chunks(range(7), 3)))   # [[0, 1, 2], [3, 4, 5], [6]]

Python 3.12 でまさにこの用途の itertools.batched が、3.10 で pairwise が追加されました。Python 3.9 では itertools.pairwise が AttributeError: module 'itertools' has no attribute 'pairwise' になり、Python 3.11 では動きましたが batched はまだありませんでした。3.9 をサポートするなら自作版を残しておきます。

実測：リスト内包表記 対 ジェネレータ式

メモリは tracemalloc で呼び出し中の追跡割り当てのピークを、時間は追跡を切った別パスで5回中の最良値を測りました。最初に追跡を有効にしたまま時間を測ったところ、同じコードが10倍以上遅くなった（42 ms → 689 ms）からです。メモリと時間は別々に測る必要があります。

"""List vs generator: tracemalloc peak and wall time (Python 3.9)."""
import time, tracemalloc

def sum_list(n):  return sum([i * i for i in range(n)])
def sum_gen(n):   return sum(i * i for i in range(n))

def peak_kb(fn, n):
    tracemalloc.start()
    fn(n)
    _, peak = tracemalloc.get_traced_memory()
    tracemalloc.stop()
    return peak / 1024

def best_ms(fn, n, repeat=5):
    best = float("inf")
    for _ in range(repeat):
        t0 = time.perf_counter()
        fn(n)
        best = min(best, time.perf_counter() - t0)
    return best * 1000

for n in (1_000_000, 10_000_000):
    for label, fn in (("list comp", sum_list), ("generator", sum_gen)):
        print(f"n={n:>10,}  {label:<10} peak={peak_kb(fn, n):>11,.1f} KB"
              f"  best of 5={best_ms(fn, n):8.1f} ms")

n= 1,000,000  list comp  peak=   39,372.4 KB  best of 5=    42.0 ms
n= 1,000,000  generator  peak=        0.7 KB  best of 5=    43.0 ms
n=10,000,000  list comp  peak=  399,378.7 KB  best of 5=   493.5 ms
n=10,000,000  generator  peak=        0.3 KB  best of 5=   516.5 ms

（1024 で割っているので出力の KB は KiB です。）2回目の実行は 40.2/43.7 ms と 503.1/522.4 ms で、メモリの値は同じでした。

• リストのメモリは要素あたり約40バイト（リスト内のポインタ8バイト＋大きな整数オブジェクト32バイト）で線形に増えます。1000万件なら、何もしないうちに 390 MiB です。
• ジェネレータのピークは n に関係なく 1 KiB 未満です。値が一度に一つしか存在しないからです。
• 速度は引き分けです。要素ごとにフレームを再開するコストのため、ジェネレータが 2〜7% 遅くなりました。ジェネレータを選ぶ理由は速度ではなく、メモリと途中終了です。

大きなファイルのストリーミング処理

実務での使いどころは、丸ごと読みたくないログファイルです。以下のスクリプトは200万行（86 MiB、seed 7）の疑似アクセスログを作り、パスごとの HTTP 500 件数をジェネレータのパイプラインと readlines()＋リスト内包表記の2通りで数えます。

"""Streaming a large log file with a generator pipeline (Python 3.9 stdlib)."""
import os, random, sys, time, tracemalloc
from collections import Counter

PATH = "access.log"

def make_log(path, n=2_000_000, seed=7):
    rng = random.Random(seed)
    paths = ["/", "/api/v1/user", "/api/v1/items", "/static/app.js", "/login"]
    codes = [200] * 90 + [304] * 5 + [404] * 3 + [500] * 2
    with open(path, "w") as f:
        for i in range(n):
            f.write(f"2026-11-12T10:{i % 60:02d}:00 GET {rng.choice(paths)} "
                    f"{rng.choice(codes)} {rng.randint(80, 9000)}ms\n")

def read_lines(path):
    with open(path) as f:
        for line in f:                 # the file object is itself a lazy iterator
            yield line.rstrip("\n")

def parse(lines):
    for line in lines:
        _ts, _method, path, code, ms = line.split(" ")
        yield path, int(code), int(ms[:-2])

def only_errors(records):
    return (r for r in records if r[1] >= 500)

def streaming(path):
    return Counter(p for p, _c, _ms in only_errors(parse(read_lines(path))))

def eager(path):
    with open(path) as f:
        lines = f.readlines()                       # whole file in memory
    records = [l.rstrip("\n").split(" ") for l in lines]
    return Counter(r[2] for r in records if int(r[3]) >= 500)

def run(fn):
    tracemalloc.start()
    res = fn(PATH)
    peak = tracemalloc.get_traced_memory()[1]
    tracemalloc.stop()
    t0 = time.perf_counter(); fn(PATH); dt = time.perf_counter() - t0
    print(f"{fn.__name__:<10} peak={peak/1024:11,.1f} KiB  time(no tracing)={dt:5.2f} s  {res.most_common(2)}")

if __name__ == "__main__":
    if not os.path.exists(PATH):
        make_log(PATH)
    print(f"file: {os.path.getsize(PATH)/2**20:.1f} MiB, Python {sys.version.split()[0]}")
    run(streaming)
    run(eager)

file: 86.0 MiB, Python 3.9.6
streaming  peak=       29.7 KiB  time(no tracing)= 1.13 s  [('/', 8057), ('/static/app.js', 8044)]
eager      peak=1,051,415.9 KiB  time(no tracing)= 3.89 s  [('/', 8057), ('/static/app.js', 8044)]

5回の実行で、ストリーミングは 1.11〜1.37 秒、一括読み込みは 3.55〜3.89 秒で、どちらも同じ 40,053 件のエラーを見つけました。86 MiB のファイルが一括版では 1 GiB の Python オブジェクトになります。1行ごとに str（ヘッダ込みで約100バイト）になり、さらに str 5個のリストになるからです。ストリーミングのパイプラインは一度に1行しか持たず、各段は数行の関数なのでリストを渡して単体テストできます。消費側が止まれば読み込みも止まります。Counter(...) を next(...) に替えれば最初のエラーまでしか読みません。メモリ上限の厳しいレンタルサーバーの cron でログを集計するなら、この差が「動くか、OOM で落ちるか」の差になります。

send() を少しだけ

ジェネレータは値を受け取ることもできます。value = yield x は x を渡して停止し、gen.send(v) は value = v として再開させます。移動平均の例です。

def running_avg():
    total, count, avg = 0.0, 0, None
    while True:
        value = yield avg
        total += value
        count += 1
        avg = total / count

avg = running_avg()
print(next(avg))          # prime it: runs to the first yield
print(avg.send(10))
print(avg.send(20))
print(avg.send(60))
fresh = running_avg()
fresh.send(5)

None
10.0
15.0
30.0
Traceback (most recent call last):
  File "~/demo/send.py", line 15, in <module>
    fresh.send(5)
TypeError: can't send non-None value to a just-started generator

最終行が典型的な間違いです。作ったばかりのジェネレータはまだ yield に到達していないので、値を受け取る場所がありません。先に next() を1回呼びます。async/await 以前の旧式コルーチンはこの仕組みの上に作られていましたが、新しいコードではメソッド一つの小さなクラスのほうが読みやすいことがほとんどです。asyncio は第5回で扱います。

どれを使うかの判断表

正直な限界とバージョンメモ

• 時間計測はすべて Python 3.9.6、Apple Silicon Mac（M1 Pro）。メモリ比較は5回中の最良値、ログファイルは5回実行の範囲。seed は固定（ログは 7）。Python 3.11 は動作とエラーメッセージの確認にだけ使い、時間計測には使っていません。
• tracemalloc が数えるのは Python のアロケータ経由のメモリだけです。RSS ではないので、インタプリタ本体やその他のバッファは含まれません。同じコードの2つの版を比べる道具であって、コンテナのメモリ上限を決める道具ではありません。
• Python 3.11 ではトレースバックの失敗した式の下に ~~~~^~~~ が付きます。それ以外のメッセージは 3.9 と 3.11 で同じでした。
• PEP 479（StopIteration → RuntimeError）は 3.7 からデフォルトです。itertools.pairwise は 3.10+（3.9 になく 3.11 にあることを確認）、itertools.batched は 3.12+ で、今回は実行していません。

シリーズ： 前回：Python 講座 2：関数・スコープ・デフォルト引数 ・ 次回：Python 講座 4：デコレータとコンテキストマネージャ。同じ「遅延評価」は C# LINQ の基礎の遅延実行としても登場し、第6回のプロファイリングと最適化でメモリと時間の計測をさらに掘り下げます。