ファイル 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() で十分だ。
