大容量ファイルをメモリに載せずに処理する（イテレータ、mmap）

GB 単位の大容量ファイルを read() で一度に読み込むと、メモリ不足でプログラムがクラッシュする。イテレータや mmap を使えば、メモリを節約しながら処理できる。

行単位で読み込む

テキストファイルは for ループで 1 行ずつ読み込める。これが最も一般的な方法だ。

# ✅ メモリ効率が良い
with open('huge_file.txt') as f:
    for line in f:
        process(line)

ファイルオブジェクトはイテレータとして動作するため、1 行分のメモリしか消費しない。

# ❌ 全体をメモリに載せる（危険）
with open('huge_file.txt') as f:
    lines = f.readlines()  # GB 単位のリストがメモリに載る
    for line in lines:
        process(line)

チャンク単位で読み込む

バイナリファイルや、行区切りでないデータはチャンク単位で読み込む。

def read_in_chunks(file_path, chunk_size=8192):
    with open(file_path, 'rb') as f:
        while chunk := f.read(chunk_size):
            yield chunk

for chunk in read_in_chunks('huge_file.bin'):
    process(chunk)

chunk_size は 4KB〜64KB 程度が一般的だ。小さすぎると I/O オーバーヘッドが増え、大きすぎるとメモリを消費する。

mmap を使う

mmap（メモリマップドファイル）を使うと、ファイルをメモリのように扱える。OS が必要な部分だけを自動的にロードするため、大容量ファイルでも効率的にアクセスできる。

import mmap

with open('huge_file.bin', 'rb') as f:
    with mmap.mmap(f.fileno(), 0, access=mmap.ACCESS_READ) as mm:
        # ファイル全体がメモリに載っているかのようにアクセス
        print(mm[0:100])  # 最初の100バイト
        print(mm[-100:])  # 最後の100バイト
        
        # 検索も可能
        pos = mm.find(b'target')

mmap の利点は以下の通りだ。

ランダムアクセスが高速（シーク不要）

OS がページング処理を行うため、実際に使う部分だけがメモリに載る

複数プロセスで共有できる

mmap で書き込み

書き込みモードで mmap を使う例を示す。

import mmap

with open('data.bin', 'r+b') as f:
    with mmap.mmap(f.fileno(), 0) as mm:
        # 特定位置に書き込み
        mm[0:5] = b'Hello'
        
        # 変更は自動的にファイルに反映される

大容量 CSV の処理

pandas で大容量 CSV を読む場合は chunksize を指定する。

import pandas as pd

for chunk in pd.read_csv('huge.csv', chunksize=10000):
    process(chunk)  # 10000行ずつ処理

メモリに収まる量の行だけを順次読み込むため、数 GB の CSV でも処理できる。

ジェネレータで処理をつなげる

複数の処理をジェネレータでつなげると、メモリ効率を維持したままパイプライン処理ができる。

def read_lines(path):
    with open(path) as f:
        for line in f:
            yield line.strip()

def filter_lines(lines, keyword):
    for line in lines:
        if keyword in line:
            yield line

def transform_lines(lines):
    for line in lines:
        yield line.upper()

# パイプライン処理（メモリに 1 行分しか載らない）
lines = read_lines('huge.txt')
filtered = filter_lines(lines, 'error')
transformed = transform_lines(filtered)

for line in transformed:
    print(line)

itertools を活用する

itertools を使うとさらに柔軟な処理ができる。

import itertools

with open('huge.txt') as f:
    # 最初の100行だけ処理
    for line in itertools.islice(f, 100):
        print(line)

import itertools

with open('huge.txt') as f:
    # 10行ずつまとめて処理
    while batch := list(itertools.islice(f, 10)):
        process_batch(batch)