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)
