Python脚本如何操作CSV大文件

本文目录导读：

1. 使用Pandas分块读取（最常用）
2. 使用Python内置csv模块
3. 使用Dask（分布式计算）
4. 按列选择性读取
5. 处理超大型文件（GB级别）
6. 性能优化技巧
7. 实际案例：合并多个大CSV文件
8. 选择建议

操作大型CSV文件时,核心原则是避免一次性将所有数据加载到内存中，以下是几种Python处理大CSV文件的方案：

使用Pandas分块读取（最常用）

import pandas as pd
# 分块读取，chunksize指定每批读取的行数
chunk_size = 10000
chunks = []
for chunk in pd.read_csv('large_file.csv', chunksize=chunk_size):
    # 处理每个分块
    # 过滤、计算、转换
    processed = chunk[chunk['column'] > 100]  # 示例过滤
    chunks.append(processed)
# 如果需要合并结果
result = pd.concat(chunks)

逐行处理（避免内存占用）

for chunk in pd.read_csv('large_file.csv', chunksize=chunk_size):
    # 处理完直接写入新文件，不保留在内存
    chunk.to_csv('output.csv', mode='a', header=False, index=False)

使用Python内置csv模块

import csv
with open('large_file.csv', 'r') as infile, open('output.csv', 'w', newline='') as outfile:
    reader = csv.reader(infile)
    writer = csv.writer(outfile)
    # 读取表头
    header = next(reader)
    writer.writerow(header)
    # 逐行处理
    for row in reader:
        # 处理逻辑
        if float(row[2]) > 100:  # 示例：筛选第三列大于100的行
            writer.writerow(row)

使用Dask（分布式计算）

import dask.dataframe as dd
# 按分区读取（自动分块）
df = dd.read_csv('large_file.csv', blocksize='100MB')
# 延迟计算
result = df[df['value'] > 100].compute()

按列选择性读取

# Pandas只读取需要的列
df = pd.read_csv('large_file.csv', usecols=['col1', 'col3', 'col5'])
# 指定数据类型以减少内存
dtype_dict = {'col1': 'int32', 'col2': 'float32'}
df = pd.read_csv('large_file.csv', dtype=dtype_dict)

处理超大型文件（GB级别）

# 方式一：生成器逐行处理
def process_large_csv(file_path):
    with open(file_path, 'r') as f:
        header = f.readline().strip().split(',')
        yield header
        for line in f:
            row = line.strip().split(',')
            # 处理每一行
            yield row
# 使用
for row in process_large_csv('huge_file.csv'):
    # 逐行处理，内存占用极低
    pass

性能优化技巧

import csv
from itertools import islice
# 1. 只读取前N行预览
preview = pd.read_csv('large.csv', nrows=100)
# 2. 跳过不需要的行
df = pd.read_csv('large.csv', skiprows=lambda x: x % 2 == 0)  # 偶数行
# 3. 多进程处理
from multiprocessing import Pool
import pandas as pd
def process_chunk(chunk_file):
    df = pd.read_csv(chunk_file)
    # 处理逻辑
    return df
# 先将大文件分割成小文件
# 然后用Pool并行处理

实际案例：合并多个大CSV文件

import pandas as pd
import glob
chunk_size = 50000
output_file = 'merged.csv'
# 写入表头
first = True
for file in glob.glob('data_*.csv'):
    for chunk in pd.read_csv(file, chunksize=chunk_size):
        chunk.to_csv(output_file, mode='a', 
                     header=first, index=False)
        first = False

选择建议

核心原则：永远不要用 pd.read_csv() 不加参数读取大文件，这会试图将所有数据加载到内存中。