Pandas 分块 CSV 聚合

示例说明

单个交易日千万行的 CSV 远超内存总量，read_csv 一次性加载必然失败。分块迭代的正确姿势是尽早聚合：每块读入后立即按维度收缩成小得多的局部汇总，内存里永远只有一块原始数据加累积中的汇总——分块大小决定内存上界，与文件总体量无关。

技术要点

• chunksize 控制单块内存上界
• 局部聚合尽早收缩数据体积
• dtype 类别化降低分组开销

代码解读

读取时只选四列并声明类别 dtype，减少解析与内存成本；循环内每块按地区与品类分组求和后暂存。合并阶段把全部局部结果拼接再做一次同维度聚合——两次聚合的键一致所以可直接相加。attrs 记录源行数供对账。

使用建议

数据倾斜时某块的分组可能仍很大，必要时二级分块。结果对账用总行数与外部计数核对。真正的超大数据集考虑 Polars 或数据库侧聚合。

代码示例

import pandas as pd

CHUNK_ROWS = 100_000


def aggregate_csv(path: str) -> pd.DataFrame:
    partials: list[pd.DataFrame] = []
    total_rows = 0

    reader = pd.read_csv(
        path,
        chunksize=CHUNK_ROWS,
        usecols=['region', 'category', 'amount', 'quantity'],
        dtype={'region': 'category', 'category': 'category'},
    )

    for chunk in reader:
        total_rows += len(chunk)
        partial = (
            chunk
            .groupby(['region', 'category'], observed=True)
            .agg(revenue=('amount', 'sum'), units=('quantity', 'sum'))
            .reset_index()
        )
        partials.append(partial)

    combined = pd.concat(partials, ignore_index=True)
    final = (
        combined
        .groupby(['region', 'category'], observed=True)
        .agg(revenue=('revenue', 'sum'), units=('units', 'sum'))
        .reset_index()
        .sort_values('revenue', ascending=False)
    )
    final.attrs['source_rows'] = total_rows
    return final


report = aggregate_csv('sales/2026-10-transactions.csv')
print('regions reported:', len(report))
print('source rows:', report.attrs['source_rows'])