PythonのGeneratorとyield：大規模データをRAMなしで処理する

1. 概要

数GBのログファイルやデータベースから数百万のレコードをストリーミングする際、「全てをリストに読み込んでから処理する」というアプローチはプログラムのRAM消費を増やし、クラッシュさせる可能性があります。この記事では、generatorとyieldがどのように機能するか、通常のリストとの比較、大規模ファイルの読み込みやデータ処理パイプラインなどの実際の問題への適用について説明します。読み終わると、generatorを正しく書く方法と、それをいつ使わないかを理解できます。

2. 従来のアプローチの問題点

例えば、5GBのログファイルを読み込んで、「ERROR」を含む行の数をカウントする必要があると仮定します：

def count_errors_bad(filepath: str) -> int:
    with open(filepath, "r") as f:
        lines = f.readlines()  # ファイル全体をRAMに読み込む
    return sum(1 for line in lines if "ERROR" in line)

f.readlines() はファイル全体をメモリ内の文字列のリストとして読み込みます。5GBのファイルの場合、Pythonはカウントを開始する前に～5GB以上のRAMを割り当てます（オブジェクトのオーバーヘッドを含む）。マシンのメモリが4GBだけの場合、プログラムは MemoryError が発生するか、スワップによるシステムハングが起こります。

コアの問題は：一度に1行ずつ処理する必要があるだけなのに、コードはマシンが同時にすべての行を保持することを強制しています。

3. Generatorとは何か？

Generatorは特別なiterableで、全ての値を計算して保存するのではなく、求められたときだけ値を生成（遅延評価）します。全体のリストを返す代わりに、generatorは1つずつ値を返し、次の値の計算は求められた時だけ行われます。

def count_errors_good(filepath: str) -> int:
    with open(filepath, "r") as f:
        return sum(1 for line in f if "ERROR" in line)

ここで、f（ファイルオブジェクト）は既にiteratorです — ループが要求する時に1行ずつ読み込み、ファイル全体をRAMに読み込みません。メモリ使用量は一度に1行だけで、ファイルが5GBでも50GBでも同じです。

4. キーワード yield：generator関数の作成方法

関数内で yield を使用することで、独自のgeneratorを書くことができます。関数が yield を含む場合、Pythonは自動的にそれをgenerator関数に変換します — 呼び出すと、すぐにコードを実行するのではなく、generatorオブジェクトを返します。

def read_large_file(filepath: str):
    with open(filepath, "r") as f:
        for line in f:
            yield line.strip()

• yield line.strip() は関数をここで一時停止し、line.strip() の値を外に返します。
• 次の next() 呼び出しのとき、関数は停止した場所から再開し、最初から再実行されません。
• 関数の状態（ローカル変数、ファイルポインタの位置）は、各 yield の間に保持されます。

試しに呼び出してみます：

gen = read_large_file("access.log")
print(type(gen))
print(next(gen))
print(next(gen))

<class 'generator'>
127.0.0.1 - GET /index.html 200
127.0.0.1 - GET /style.css 200

注意：gen = read_large_file(...) の行は、ファイルを開かず、何も読み込みません。内部のコードは、next(gen) を初めて呼び出すときだけ実行されます。

5. RAMの比較：generator vs list

違いをはっきり見るために、sys モジュールを使用してオブジェクトのサイズを測定します：

import sys

def squares_list(n: int) -> list[int]:
    return [i * i for i in range(n)]

def squares_gen(n: int):
    for i in range(n):
        yield i * i

big_list = squares_list(1_000_000)
big_gen = squares_gen(1_000_000)

print(f"List size: {sys.getsizeof(big_list):,} bytes")
print(f"Generator size: {sys.getsizeof(big_gen):,} bytes")

List size: 8,448,728 bytes
Generator size: 208 bytes

100万要素のリストはほぼ8.5MBを占有しますが、generatorはわずか208バイトのオブジェクトです — 値を保存しないため、必要なときに次の値を計算するための状態のみを保存します。

注意： Generatorは一度だけ走査可能です。走査を完了した後、next() を呼び出すと、StopIteration が発生します。複数回走査する必要がある場合は、関数を再度呼び出して新しいgeneratorを作成する必要があります。

6. Generator式：よりコンパクトな構文

リスト内包表記と似ていますが、角括弧の代わりに括弧を使用します。generator式は関数を個別に定義する必要はありません：

numbers = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]

even_squares_gen = (n * n for n in numbers if n % 2 == 0)
print(sum(even_squares_gen))

220

括弧 () を [] の代わりに使用すると、Pythonはすぐにリストを作成する代わりにgeneratorを作成します。この方法は、値を一度だけ使用する場合に便利です。例えば、sum(), max(), any() に直接渡す場合です。

7. Generatorを使用したデータ処理パイプラインの構築

Generatorの強みは、複数の処理ステップをチェーンすることができ、中間のリストを作成しないことです。例えば、注文CSVを読み込む → status = "PAID" の注文をフィルタリング → amount の合計を計算する必要があると仮定します。

import csv

def read_orders(filepath: str):
    with open(filepath, newline="") as f:
        reader = csv.DictReader(f)
        for row in reader:
            yield row

def filter_paid(orders):
    for order in orders:
        if order["status"] == "PAID":
            yield order

def extract_amount(orders):
    for order in orders:
        yield float(order["amount"])

def total_paid_amount(filepath: str) -> float:
    orders = read_orders(filepath)
    paid_orders = filter_paid(orders)
    amounts = extract_amount(paid_orders)
    return sum(amounts)

各関数 read_orders, filter_paid, extract_amount はgeneratorです — 全データを一度に処理するのではなく、パイプラインにチェーンされます。sum(amounts) が next() を呼び出すと、値は各ステップを通して引き出されます：1行のCSVを読み込む → ステータスを確認 → floatに変換 → 累積します。いかなる時点でも、CSVファイルが10行でも1000万行でも、メモリ内には1つのレコードだけが存在します。

7.1. pytestでパイプラインをテストする

import pytest
from pathlib import Path

@pytest.fixture
def orders_csv(tmp_path: Path) -> str:
    content = (
        "order_id,amount,status\n"
        "1,100.0,PAID\n"
        "2,50.0,PENDING\n"
        "3,200.0,PAID\n"
    )
    file_path = tmp_path / "orders.csv"
    file_path.write_text(content)
    return str(file_path)

def test_total_paid_amount(orders_csv):
    result = total_paid_amount(orders_csv)
    assert result == pytest.approx(300.0)

tmp_path はpytestの組み込みフィクスチャで、各テストのための一時ディレクトリを作成します。このテストは3つの注文を含む偽のCSVファイルを作成し、そのうち2つだけが PAID であり、合計が300.0であることを確認します。

8. yield from：別のgeneratorへの委譲

generatorが別のgeneratorから結果を呼び出してネストする必要がある場合、yield from はループを手動で記述する代わりに、よりコンパクトなコードを実現します。

def read_file_lines(filepath: str):
    with open(filepath) as f:
        for line in f:
            yield line.strip()

def read_multiple_files(filepaths: list[str]):
    for filepath in filepaths:
        yield from read_file_lines(filepath)

yield from read_file_lines(filepath) は for line in read_file_lines(filepath): yield line と同等ですが、よりコンパクトで、generator副ルーチンが return 値を持つ場合を正しく処理します。関数 read_multiple_files により、複数のログファイルを連続したストリームのように走査でき、リストをマージする必要がなくなります。

9. send() を使用してgeneratorに値を送信する

Generatorは値を外に送信するだけではありません — send() を使用して内部に値を送信し、単純なcoroutineに変えることができます。

def running_average():
    total = 0.0
    count = 0
    average = None
    while True:
        value = yield average
        total += value
        count += 1
        average = total / count

avg_gen = running_average()
next(avg_gen)              # generatorを初期化、最初の yield に到達
print(avg_gen.send(10))    # 10 を送信、average = 10/1
print(avg_gen.send(20))    # 20 を送信、average = 30/2
print(avg_gen.send(30))    # 30 を送信、average = 60/3

10.0
15.0
20.0

value = yield average 行は2つのことを行います：average を外に返すと同時に、send() を通じて送信された値を待ち受けて、value に割り当てます。generatorが実行を開始していない状態で send() を呼び出すと、Pythonは TypeError を発生させるため、最初に next(avg_gen) を呼び出す必要があります。

注意： send() の使用は、通常のgeneratorより読みやすくしてしまいます。実際のところ、asyncio のようなフレームワークはこのメカニズムを抽象化しており、ライブラリを書いている場合を除き、send() を自分で書くことはほぼありません。

10. Generatorが向かない場合

Generatorが常にリストより優れているわけではありません：

• ランダムアクセスまたは複数回の走査が必要： generatorは gen[5] をサポートしず、最初からやり直すこともできません。len(), スライシング、または2～3回のデータ走査が必要な場合、リストがより適切です。
• データが小さく、RAMについて心配する必要がない： 数千要素の場合、generatorの状態管理のオーバーヘッドは、Cレベルで最適化されたリスト内包表記より遅いことがあります。
• 並列処理または複数のworkerに仕事を分割する必要がある： リストは小さなチャンクに分割して、multiprocessingに送信するのが簡単です；generatorは分割する前に一部を「実体化」する必要があります。
• デバッグがより難しい： リストを出力するとすべての値が見えます；generatorを出力するのは <generator object ... at 0x...> だけを見るため、値を見るには走査する必要があります（走査後、generatorは「枯渇」して再利用できません）。

不確定な場合の簡単なルール：データがRAMに収まり、複数回使用する必要がある場合はリストを使用；データが大きく、順序に処理する必要がある場合、またはストリーミングパイプラインを構築する場合はgeneratorを使用します。

11. 結論

GeneratorとYieldは正しい問題を解決します：遅延評価のメカニズムのおかげで、全データをRAMに読み込まずに大規模データを処理します — 要求されるまで値を計算します。generatorの書き方、generator式、yield from を使用したgeneratorのチェーン、send() を使用したgeneratorへのデータ送信方法を学びました。ただし忘れずに：generatorはメモリを節約しますが、走査を繰り返したり、ランダムアクセスしたり、コードが読みにくくなる可能性があります。問題に対して正しい道具を選んでください。すべての場所でgeneratorを使用するのではなく、「RAMを節約する」という理由だけで。