大量のデータを処理する際、すべてをメモリに読み込むとリソースを圧迫します。イテレータとジェネレータを使えば、必要な時に必要な分だけデータを生成する「遅延評価」が可能になります。
イテレータの基本
イテラブルとイテレータ
# イテラブル（Iterable）: __iter__を持つオブジェクト
numbers = [1, 2, 3]
# イテレータ（Iterator）: __next__を持つオブジェクト
iterator = iter(numbers)
print(next(iterator)) # 1
print(next(iterator)) # 2
print(next(iterator)) # 3
print(next(iterator)) # StopIteration例外
forループの内部動作
# forループは内部的にこう動作する
numbers = [1, 2, 3]
iterator = iter(numbers)
while True:
try:
item = next(iterator)
print(item)
except StopIteration:
break
カスタムイテレータ
class CountDown:
def __init__(self, start: int):
self.current = start
def __iter__(self):
return self
def __next__(self) -> int:
if self.current <= 0:
raise StopIteration
self.current -= 1
return self.current + 1
for num in CountDown(5):
print(num) # 5, 4, 3, 2, 1
ジェネレータ関数
yieldキーワードを使用した関数は、ジェネレータを返します：
def countdown(n: int):
while n > 0:
yield n
n -= 1
# ジェネレータオブジェクトを取得
gen = countdown(5)
print(type(gen)) # <class 'generator'>
# イテレート
for num in gen:
print(num) # 5, 4, 3, 2, 1
yieldの動作
def simple_generator():
print("Start")
yield 1
print("After first yield")
yield 2
print("After second yield")
yield 3
print("End")
gen = simple_generator()
print(next(gen)) # "Start" を出力後、1を返す
print(next(gen)) # "After first yield" を出力後、2を返す
print(next(gen)) # "After second yield" を出力後、3を返す
# next(gen) # "End" を出力後、StopIteration
メモリ効率の比較
import sys
# リスト: すべてをメモリに保持
def get_squares_list(n: int) -> list[int]:
return [x * x for x in range(n)]
# ジェネレータ: 必要な時に生成
def get_squares_gen(n: int):
for x in range(n):
yield x * x
n = 1_000_000
list_result = get_squares_list(n)
gen_result = get_squares_gen(n)
print(f"List size: {sys.getsizeof(list_result):,} bytes") # ~8MB
print(f"Generator size: {sys.getsizeof(gen_result):,} bytes") # ~200 bytes
ジェネレータ式
リスト内包表記をジェネレータに：
# リスト内包表記（即座に全要素を生成）
squares_list = [x * x for x in range(1000)]
# ジェネレータ式（遅延評価）
squares_gen = (x * x for x in range(1000))
# 使用例
total = sum(x * x for x in range(1000000)) # メモリ効率が良い
いつジェネレータ式を使うか
flowchart TD
Q1{"すべての結果が<br/>必要？"}
Q2{"結果を複数回<br/>使う？"}
Q3{"データサイズが<br/>大きい？"}
Q1 -->|No| Gen["ジェネレータ式"]
Q1 -->|Yes| Q2
Q2 -->|Yes| List["リスト内包表記"]
Q2 -->|No| Q3
Q3 -->|Yes| Gen
Q3 -->|No| List
style Gen fill:#22c55e,color:#fff
style List fill:#3b82f6,color:#fff
yield from
サブジェネレータに委譲：
def flatten(nested_list):
for item in nested_list:
if isinstance(item, list):
yield from flatten(item) # 再帰的に委譲
else:
yield item
nested = [1, [2, 3, [4, 5]], 6, [7, 8]]
print(list(flatten(nested))) # [1, 2, 3, 4, 5, 6, 7, 8]
yield fromとforの違い
# forを使った場合
def chain_with_for(*iterables):
for it in iterables:
for item in it:
yield item
# yield fromを使った場合（より簡潔）
def chain_with_yield_from(*iterables):
for it in iterables:
yield from it
# 同じ結果
result1 = list(chain_with_for([1, 2], [3, 4])) # [1, 2, 3, 4]
result2 = list(chain_with_yield_from([1, 2], [3, 4])) # [1, 2, 3, 4]
itertools
標準ライブラリの強力なイテレータツール：
無限イテレータ
from itertools import count, cycle, repeat
# count: 無限カウンタ
for i in count(10, 2): # 10から2ずつ
if i > 20:
break
print(i) # 10, 12, 14, 16, 18, 20
# cycle: 無限ループ
colors = cycle(["red", "green", "blue"])
for _, color in zip(range(5), colors):
print(color) # red, green, blue, red, green
# repeat: 同じ値を繰り返し
for x in repeat("hello", 3):
print(x) # hello, hello, hello
組み合わせ
from itertools import chain, product, permutations, combinations
# chain: イテラブルを連結
print(list(chain([1, 2], [3, 4], [5]))) # [1, 2, 3, 4, 5]
# product: デカルト積
print(list(product("AB", [1, 2])))
# [('A', 1), ('A', 2), ('B', 1), ('B', 2)]
# permutations: 順列
print(list(permutations("ABC", 2)))
# [('A', 'B'), ('A', 'C'), ('B', 'A'), ('B', 'C'), ('C', 'A'), ('C', 'B')]
# combinations: 組み合わせ
print(list(combinations("ABC", 2)))
# [('A', 'B'), ('A', 'C'), ('B', 'C')]
フィルタリングと変換
from itertools import filterfalse, takewhile, dropwhile, islice
numbers = [1, 4, 6, 4, 4, 2, 8, 3]
# filterfalse: 条件がFalseの要素
print(list(filterfalse(lambda x: x % 2, numbers))) # [4, 6, 4, 4, 2, 8]
# takewhile: 条件がTrueの間取得
print(list(takewhile(lambda x: x < 5, numbers))) # [1, 4]
# dropwhile: 条件がTrueの間スキップ
print(list(dropwhile(lambda x: x < 5, numbers))) # [6, 4, 4, 2, 8, 3]
# islice: スライス操作
print(list(islice(numbers, 2, 5))) # [6, 4, 4]
グループ化
from itertools import groupby
data = [
("A", 1), ("A", 2), ("B", 3), ("B", 4), ("A", 5)
]
# 連続した同じキーでグループ化（ソート済みが前提）
sorted_data = sorted(data, key=lambda x: x[0])
for key, group in groupby(sorted_data, key=lambda x: x[0]):
print(f"{key}: {list(group)}")
# A: [('A', 1), ('A', 2), ('A', 5)]
# B: [('B', 3), ('B', 4)]
実践的なパターン
ファイルの遅延読み込み
def read_large_file(path: str):
"""大きなファイルを1行ずつ処理"""
with open(path, 'r') as f:
for line in f:
yield line.strip()
# メモリ効率よく処理
for line in read_large_file("huge_file.txt"):
if "error" in line.lower():
print(line)
パイプライン処理
def read_data(path: str):
with open(path) as f:
for line in f:
yield line.strip()
def parse_json(lines):
import json
for line in lines:
yield json.loads(line)
def filter_active(records):
for record in records:
if record.get("active"):
yield record
def extract_names(records):
for record in records:
yield record["name"]
# パイプラインを構築
pipeline = extract_names(
filter_active(
parse_json(
read_data("users.jsonl")
)
)
)
# 遅延評価で効率的に処理
for name in pipeline:
print(name)
バッチ処理
from itertools import islice
def batched(iterable, n: int):
"""イテラブルをn個ずつのバッチに分割"""
iterator = iter(iterable)
while batch := list(islice(iterator, n)):
yield batch
# 1000件のデータを100件ずつ処理
items = range(1000)
for batch in batched(items, 100):
process_batch(batch) # 100件ずつ処理
無限シーケンス
def fibonacci():
"""無限フィボナッチ数列"""
a, b = 0, 1
while True:
yield a
a, b = b, a + b
# 最初の10個を取得
from itertools import islice
print(list(islice(fibonacci(), 10)))
# [0, 1, 1, 2, 3, 5, 8, 13, 21, 34]
send()とジェネレータの双方向通信
def accumulator():
total = 0
while True:
value = yield total
if value is not None:
total += value
acc = accumulator()
next(acc) # ジェネレータを起動
print(acc.send(10)) # 10
print(acc.send(20)) # 30
print(acc.send(15)) # 45
まとめ
イテレータとジェネレータは、メモリ効率と処理効率を両立させる強力なツールです：
| 概念 | 説明 | ユースケース |
|---|---|---|
| イテレータ | __next__で値を順次取得 |
カスタム反復処理 |
| ジェネレータ関数 | yieldで値を生成 |
遅延評価、無限シーケンス |
| ジェネレータ式 | (expr for x in iter) |
ワンライナーの遅延評価 |
yield from |
サブジェネレータに委譲 | ネストした構造の平坦化 |
| itertools | 標準ライブラリのツール | 組み合わせ、フィルタリング |
主要な原則：
- 遅延評価を活用: 大量データはジェネレータで処理
- パイプラインを構築: ジェネレータをチェーンして効率化
- itertoolsを知る: 車輪の再発明を避ける
- メモリを意識: リストが必要な時以外はジェネレータ
ジェネレータは「必要な時に必要な分だけ」というPythonの効率的なデータ処理の要です。