Pandas 大容量ファイル処理（chunksize）

大きなデータファイルを処理する際、メモリが不足して一度にすべてのデータを読み込めないことがあります。Pandas はチャンク読み込み機能を提供しており、データをバッチ単位で処理できます。

CSV のチャンク読み込み

chunksize パラメータ

例

import pandas as pd
import io

# 大ファイルの内容をシミュレート（実際の使用時はファイルパスに置き換え）
data = """id,value
1,100
2,200
3,300
4,400
5,500
6,600
7,700
8,800
9,900
10,1000
"""

# chunksize を使用してチャンク単位で読み込む
chunks = pd.read_csv(io.StringIO(data), chunksize=3)

print("チャンク処理：")
for i, chunk in enumerate(chunks):
    print(f"\n"チャンク {i+1}:")
    print(chunk)

集計処理

例

import pandas as pd
import io
import numpy as np

# 大ファイルのシミュレート
data = "value\n" + "\n".join([str(i) for i in range(1, 1001)])

# チャンクごとに合計を計算
total = 0
count = 0

for chunk in pd.read_csv(io.StringIO(data), chunksize=100):
    total += chunk["value"].sum()
    count += len(chunk)

print(f"総行数: {count}")
print(f"合計: {total}")
print(f"平均値: {total / count}")

増分処理

フィルタリング

例

import pandas as pd
import io

# データのシミュレート
data = """id,value,category
1,100,A
2,200,B
3,150,A
4,300,C
5,250,B
6,180,A
"""

# 特定の条件に一致するデータをフィルタリング
filtered_chunks = []

for chunk in pd.read_csv(io.StringIO(data), chunksize=2):
    filtered = chunk[chunk["category"] == "A"]
    if len(filtered) > 0:
        filtered_chunks.append(filtered)

result = pd.concat(filtered_chunks)
print("category='A' のデータをフィルタリング：")
print(result)

マルチプロセス並列処理

例

import pandas as pd
import io
from concurrent.futures import ProcessPoolExecutor

# 並列処理（CPU バウンドなタスク向け）
def process_chunk(chunk):
    """単一のデータチャンクを処理"""
    return chunk["value"].sum()

# データのシミュレート
data = "value\n" + "\n".join([str(i) for i in range(1, 101)])

# データチャンクを準備
chunks = list(pd.read_csv(io.StringIO(data), chunksize=10))

# シリアル処理
total = sum(process_chunk(chunk) for chunk in chunks)
print(f"シリアル処理の合計: {total}")

JSON 大容量ファイルの処理

例

import pandas as pd
import json
import io

# JSON Lines 形式のデータをシミュレート
jsonl_data = '\n'.join([
    json.dumps({"id": i, "value": i * 10})
    for i in range(1, 101)
])

# JSON Lines をチャンク単位で読み込む
chunks = []
chunk_size = 20

for chunk in pd.read_json(io.StringIO(jsonl_data), lines=True, chunksize=chunk_size):
    chunks.append(chunk)
    if len(chunks) >= 3:  # 例として最初の3チャンクのみ処理
        break

result = pd.concat(chunks)
print(f"{len(result)} 行のデータを読み込みました")
print(result.head())

チャンク処理によりメモリ使用量を大幅に削減できますが、処理時間は増加します。メモリと時間のバランスを取るために、chunk サイズを適切に選択してください。

その他の拡張