如何优化 Python 处理大型 CSV 文件时的内存使用

import pandas as pd

# 预先定义高效 dtype 映射（根据实际字段调整）

dtypes = {

    status: category,

    region: category,

    user_id: uint32,      # 若最大值 < 4.3B

    score: float32,

    timestamp: int64      # 或使用 pd.to_datetime(..., cache=True) 延迟解析

}

df = pd.read_csv(large_input.csv, dtype=dtypes)

print(df.info(memory_usage=deep))  # 对比优化前后内存占用

def process_large_csv_chunked(input_file, output_file, chunk_size=50000):

    first_chunk = True

    for chunk in pd.read_csv(input_file, chunksize=chunk_size, dtype=dtypes):

        # 在 chunk 上执行轻量级转换（避免 .copy() 或复杂 apply）

        filtered_chunk = chunk[chunk[value] > 100].copy()

        filtered_chunk[new_col] = filtered_chunk[old_col].str.upper()

        # 首次写入含 header，后续追加不写 header

        filtered_chunk.to_csv(

            output_file,

            mode=w if first_chunk else a,

            header=first_chunk,

            index=False

        )

        first_chunk = False

        # 可选：显式删除 chunk 引用，协助 GC

        del chunk, filtered_chunk

process_large_csv_chunked(large_input.csv, transformed_output.csv)

import polars as pl

# 惰性读取（不立即加载）

lf = pl.scan_csv(large_input.csv, dtypes={status: pl.Categorical})

result = (

    lf.filter(pl.col(value) > 100)

      .with_columns(pl.col(name).str.to_uppercase().alias(NAME))

      .sink_csv(transformed_output.csv)  # 直接写入磁盘，不返回 DataFrame

)