大容量ファイルのチャンク読み込み

ファイルが RAM を超える場合

import pandas as pd
import numpy as np

# Simulate a large CSV by writing one
np.random.seed(0)
sample = pd.DataFrame({
    'date': pd.date_range('2023-01-01', periods=100000, freq='h'),
    'region': np.random.choice(['North','South','East','West'], 100000),
    'sales': np.random.randint(100, 1000, 100000)
})
sample.to_csv('/tmp/large_sales.csv', index=False)
print(f'File size: {pd.io.common.get_handle("/tmp/large_sales.csv", "r").handle.seek(0, 2)/1e6:.1f} MB... (simulated)')
print('Rows:', len(sample))

read_csv の chunksize パラメータ

import pandas as pd

# Read file in chunks of 25,000 rows
chunk_iter = pd.read_csv('/tmp/large_sales.csv', chunksize=25000)

# Peek at the first chunk
first_chunk = next(chunk_iter)
print('First chunk shape:', first_chunk.shape)
print(first_chunk.head(3))

チャンクの反復処理

import pandas as pd

total_rows = 0
total_sales = 0.0
chunk_count = 0

for chunk in pd.read_csv('/tmp/large_sales.csv', chunksize=25000):
    total_rows += len(chunk)
    total_sales += chunk['sales'].sum()
    chunk_count += 1

print(f'Chunks processed: {chunk_count}')
print(f'Total rows: {total_rows:,}')
print(f'Total sales: {total_sales:,.0f}')
print(f'Avg sales per row: {total_sales/total_rows:.2f}')

チャンク処理中の行のフィルタリング

import pandas as pd

filtered_chunks = []

for chunk in pd.read_csv('/tmp/large_sales.csv', chunksize=25000):
    # Keep only North region rows
    north = chunk[chunk['region'] == 'North']
    if len(north) > 0:
        filtered_chunks.append(north)

north_df = pd.concat(filtered_chunks, ignore_index=True)
print(f'North region rows: {len(north_df):,}')
print(f'North total sales: {north_df["sales"].sum():,.0f}')

GroupBy の逐次集計

import pandas as pd

partials = []

for chunk in pd.read_csv('/tmp/large_sales.csv', chunksize=25000):
    # Compute sum and count per region in this chunk
    partial = chunk.groupby('region')['sales'].agg(['sum', 'count'])
    partials.append(partial)

# Combine all partial results
combined = pd.concat(partials).groupby(level=0).sum()
combined['mean'] = combined['sum'] / combined['count']

print('Regional aggregation (chunked):')
print(combined.round(2))

読み込み時に指定するメモリ効率のよい型

import pandas as pd

specified_dtypes = {
    'region': 'category',
    'sales': 'int32'
}

total_sales = 0
for chunk in pd.read_csv(
    '/tmp/large_sales.csv',
    chunksize=25000,
    dtype=specified_dtypes,
    parse_dates=['date']
):
    total_sales += chunk['sales'].sum()
    # Only 25k rows * (category + int32 + datetime) in RAM at once

print(f'Total sales (memory-efficient): {total_sales:,.0f}')

結果を逐次書き込む

import pandas as pd
import os

output_path = '/tmp/filtered_output.csv'

# Remove previous output if it exists
if os.path.exists(output_path):
    os.remove(output_path)

first_chunk = True
for chunk in pd.read_csv('/tmp/large_sales.csv', chunksize=25000):
    # Process: keep only high-value rows
    processed = chunk[chunk['sales'] > 800].copy()
    
    # Write: header only on first chunk, append thereafter
    processed.to_csv(output_path,
                     mode='a',
                     header=first_chunk,
                     index=False)
    first_chunk = False

result = pd.read_csv(output_path)
print(f'High-value rows written: {len(result):,}')
print(f'Average sales (>800): {result["sales"].mean():.1f}')

ドロップインの代替手段としての Dask の紹介

# pip install dask
# import dask.dataframe as dd

# Read the large CSV as a Dask DataFrame (lazy — no data loaded yet)
# ddf = dd.read_csv('/tmp/large_sales.csv')

# Operations are lazy — no computation happens until .compute()
# result = ddf.groupby('region')['sales'].mean().compute()
# print(result)

# Key Dask advantages:
# - Automatic chunking (no manual chunksize loops)
# - Parallel execution (multi-core)
# - Familiar Pandas API
# - Works with files larger than RAM

print('Dask extends Pandas to datasets larger than RAM with minimal API changes.')

チャンクサイズの選択

import pandas as pd
import timeit

# Benchmark different chunk sizes
for chunksize in [10000, 50000, 100000]:
    t = timeit.timeit(
        lambda: sum(chunk['sales'].sum()
                    for chunk in pd.read_csv('/tmp/large_sales.csv', chunksize=chunksize)),
        number=3
    )
    print(f'chunksize={chunksize:>7,}: {t/3:.3f}s per pass')

大規模データにおける Parquet と CSV の比較

import pandas as pd

# Convert our CSV to Parquet (do this once)
df = pd.read_csv('/tmp/large_sales.csv', parse_dates=['date'])
df['region'] = df['region'].astype('category')
df['sales'] = df['sales'].astype('int32')
df.to_parquet('/tmp/large_sales.parquet', index=False)

# Now read only the columns needed — much faster than CSV
sales_by_region = pd.read_parquet(
    '/tmp/large_sales.parquet',
    columns=['region', 'sales']
)
print('Parquet read result:')
print(sales_by_region.groupby('region')['sales'].mean().round(1))
print('\ndtypes preserved:', sales_by_region.dtypes.to_dict())

チャンク処理パイプラインの全体パターン

理解度チェック

レッスンのまとめ