【Python】ジェネレータによるメモリ最適化とストリーム設計

Pythonでプログラムを書いていて、大量データを扱う処理を書くとき、
次のような問題に直面することがある。

• メモリ使用量が急増する
• 処理開始まで時間がかかる
• ファイル全体を読み込む必要がある

これらを解決するための仕組みが「ジェネレータ」である。

ジェネレータは単なる構文ではなく、
設計レベルの問題を解決するための道具である。

本記事では、ジェネレータを実務で有効活用するための考え方を

• メモリ効率の最適化
• ストリーム処理

の観点から整理する。

メモリ効率の最適化

大規模データ処理において、最大の制約はメモリになる

CPUはスケールしやすいし、ストレージも拡張可能だが メモリは一度に確保できる上限がある。

例えば、

• 100万件 → 数十MB
• 1000万件 → 数百MB
• 1億件 → 数GB

というように、データ件数に比例してメモリ消費が増加する設計では破綻する。

つまり大規模データ処理においてはメモリを最適化することが重要になる。

即時評価（リスト内包表記）の場合

Copy
def calculate_total(data):
    total = 0
    for x in data:
        total += x
    return total

def main():
    # 1000万件を想定
    numbers = [x for x in range(10_000_000)]
    return calculate_total(numbers)

大規模データに対して、即時評価（リスト内包表記）の場合
動作としては

• rangeをすべて展開
• 1000万件のリストを作成
• その後に関数を実行

という流れになる。

calculate_total() が1件ずつ処理するにもかかわらず、
呼び出し前に全件をメモリへ展開している。

この場合、メモリ使用量が大きくなる。

遅延評価（ジェネレータ）の場合

Copy
def calculate_total(data):
    total = 0
    for x in data:
        total += x
    return total

def main():
    # ジェネレータ式で渡す
    return calculate_total(x for x in range(10_000_000))

大規模データに対して、遅延評価の場合、動作としては

• ジェネレータ式で書く →　ジェネレータを生成しただけ
• リストは生成されない
• calculate_total() が値を要求するたびに1件ずつ生成
• 生成された値は、その都度呼び出し元へ返される（全件を内部に保持しない）

となり、ジェネレータを使うことで
結果を全件メモリに保持せず、必要な分だけ順次生成する設計になる。

ジェネレータが保持するのは

• 現在の実行位置
• ローカル変数
• イテレーション状態

のみであり
データ件数に比例してメモリ使用量が増加する設計を避けることができる。
※ただし huge_dataset自体がリストである場合、入力はすでにメモリに展開されている。 真に大規模データ処理を行う場合は、入力側もストリーム（ファイル、DBカーソルなど）で扱う必要がある。

実務での使用例

悪い例としては

Copy
def calculate_total(data):
    total = 0
    for x in data:
        total += x
    return total

def main():
    # いったんジェネレータを作る
    gen = (x for x in range(10_000_000))

    # しかし結局 list に展開してしまう（ここでメモリを使う）
    numbers = list(gen)

    # 以降は list を渡しているので即時評価と同じ状態
    return calculate_total(numbers)

• ジェネレータを使っても最終的にlist化すれば意味がない。 -「ジェネレータを作ったかどうか」ではなく「全件をどこかで list 化していないか」が重要になる。

良い例としては

Copy
gen = (transform(x) for x in huge_dataset)
filtered = (x for x in gen if condition(x))
lst = list(filtered)

のように多段で処理してる場合、ジェネレータを使っていれば

• 中間リストは一切作られない
• 最終listだけ生成される

もし途中でリストを作っていたら

• 中間リスト
• 最終リスト

の2倍のメモリが必要になる。

また、途中で打ち切る場合

Copy
next(x for x in huge_dataset if x > 1000)

• 条件を満たすまでしか生成しない
• 全件評価しない

のでメモリ効率が良い。 list内包表記は全件評価される。

ストリーム処理

大規模データ処理では「入力をどう持つのか」も大きな問題となる。
この時に有効なのがストリーム処理である。

ストリームとはデータが連続的に流れてくる構造を指す
具体的には

• ログファイル
• ネットワーク受信データ
• メッセージキュー
• DBカーソル

がある。

これらに共通するのは、データが「塊」として最初から全部揃っているのではなく、順次到着・取得できる点にある。
そのため「全件をリストに読み込む」のではなく「1件ずつ取り出して処理する」設計が自然になる

例えば、読み取り → フィルタ → 変換 → 出力を「流れ」として構築できる。

Copy
def read_lines(path):
    # ファイルを開く（with により自動クローズ）
    with open(path, encoding="utf-8") as f:
        # ファイルオブジェクトはイテレータ
        # 1行ずつ読み込まれる（全件メモリに載せない）
        for line in f:
            # 1行ずつ呼び出し元へ返す
            yield line

def parse(line):
    # 行末の改行や前後の空白を除去
    return line.strip()

def filter_valid(x):
    # 空文字列は除外
    return x != ""

def pipeline(path):
    # read_lines もジェネレータ
    # line は1行ずつ供給される
    for line in read_lines(path):

        # 1行を加工
        x = parse(line)

        # 条件に合うものだけ通す
        if filter_valid(x):

            # 呼び出し元へ1件返す
            # ここで一旦処理は停止する（次の next まで）
            yield x

このように、ジェネレータを組み合わせることで「全件保持」ではなく「逐次処理」になる。
ジェネレータはストリーム設計と相性が良い理由はここにある。

まとめ

ジェネレータの本質は「遅延評価」にある。

リストが結果を保持する構造であるのに対し、
ジェネレータは値を必要な分だけ順次生成する構造である。

大規模データ処理では、

• 全件をメモリに展開しない
• 中間リストを作らない
• 入力もストリームとして扱う

といった設計が重要になる。

重要なのは「ジェネレータを使うこと」ではなく、
全件保持型の設計になっていないかを意識することである。

ジェネレータは、そのための有効な道具である。