ファイル I/O のパフォーマンスを改善するテクニック

ファイル I/O はプログラムのボトルネックになりやすい。適切なテクニックを使えば、読み書きの速度を大幅に改善できる。

バッファサイズを調整する

open() の buffering パラメータでバッファサイズを変更できる。

# デフォルト（システム依存、通常は 4KB〜8KB）
with open('data.txt') as f:
    content = f.read()

# バッファサイズを 1MB に設定
with open('data.txt', buffering=1024*1024) as f:
    content = f.read()

大きなファイルを読む場合、バッファサイズを大きくすると I/O 回数が減り、高速化できることがある。

適切なチャンクサイズで読み込む

チャンク単位で読む場合、サイズによって速度が変わる。

import time

def read_file_chunked(filepath, chunk_size):
    with open(filepath, 'rb') as f:
        while f.read(chunk_size):
            pass

# チャンクサイズ別のベンチマーク（例）
for size in [1024, 8192, 65536, 1024*1024]:
    start = time.time()
    read_file_chunked('large_file.bin', size)
    print(f'{size:>10} bytes: {time.time() - start:.3f}s')

一般的には 8KB〜64KB が良いバランスだ。小さすぎると syscall オーバーヘッドが増え、大きすぎるとメモリを消費する。

readlines() より for ループ

readlines() は全行をリストとして読み込むため、メモリを大量に消費し、最初の行を処理するまでの待ち時間も長い。

# ❌ 遅い（全行をメモリに読み込む）
with open('huge.txt') as f:
    lines = f.readlines()
    for line in lines:
        process(line)

# ✅ 速い（1行ずつ読み込む）
with open('huge.txt') as f:
    for line in f:
        process(line)

バイナリモードを使う

テキストモードはエンコーディング変換のオーバーヘッドがある。エンコーディング処理が不要なら、バイナリモードが速い。

# テキストモード（エンコーディング変換あり）
with open('data.txt', 'r', encoding='utf-8') as f:
    content = f.read()

# バイナリモード（変換なし）
with open('data.txt', 'rb') as f:
    content = f.read()

writelines() を使う

複数行を書き込む場合、writelines() は効率的だ。

lines = ['line1\n', 'line2\n', 'line3\n']

# ❌ 遅い（書き込みを複数回呼ぶ）
with open('output.txt', 'w') as f:
    for line in lines:
        f.write(line)

# ✅ 速い（一度に書き込む）
with open('output.txt', 'w') as f:
    f.writelines(lines)

書き込みをまとめる

小さな書き込みを何度も行うより、まとめて書き込む方が速い。

# ❌ 遅い（細かい書き込み）
with open('output.txt', 'w') as f:
    for i in range(100000):
        f.write(f'line {i}\n')

# ✅ 速い（まとめて書き込み）
lines = [f'line {i}\n' for i in range(100000)]
with open('output.txt', 'w') as f:
    f.write(''.join(lines))

mmap を使う

mmap はランダムアクセスが多い場合に特に効果的だ。

import mmap

# 通常の読み込み
with open('data.bin', 'rb') as f:
    f.seek(1000000)
    chunk1 = f.read(1000)
    f.seek(2000000)
    chunk2 = f.read(1000)

# mmap（ランダムアクセスが速い）
with open('data.bin', 'rb') as f:
    with mmap.mmap(f.fileno(), 0, access=mmap.ACCESS_READ) as mm:
        chunk1 = mm[1000000:1001000]
        chunk2 = mm[2000000:2001000]

非同期 I/O を使う

aiofiles を使うと、I/O 待ち時間に他の処理を行える。

pip install aiofiles

import asyncio
import aiofiles

async def read_file(filepath):
    async with aiofiles.open(filepath) as f:
        return await f.read()

async def main():
    # 複数ファイルを並行して読む
    tasks = [read_file(f'file{i}.txt') for i in range(10)]
    results = await asyncio.gather(*tasks)

asyncio.run(main())

os.read() / os.write() を使う

最低レベルの I/O 関数は Python のオーバーヘッドが少ない。

import os

fd = os.open('data.bin', os.O_RDONLY)
try:
    data = os.read(fd, 1024*1024)
finally:
    os.close(fd)

ただし、使い勝手が悪いため、通常は open() で十分だ。