如何加速Pandas将大CSV分块写入Excel的处理速度？

问题背景 #

我需要将一个约100万行、177列的CSV文件通过分块方式追加写入单个XLSX文件完成格式转换，当前使用的代码如下：

import pandas as pd
import openpyxl
import timeit
import xlsxwriter
import os
import numpy as np

def process_csv_files(csv_file_path, excel_base_path, chunk_size):
    try:
        print(f'csv_file_path : {csv_file_path}')
        print(f'excel_base_path : {excel_base_path}')
        print(f'chunk_size : {chunk_size}')
        base_file_name, _ = os.path.splitext(os.path.basename(csv_file_path))
        excel_file_path = f'{base_file_name}.xlsx'
        excel_file_path = os.path.join(excel_base_path, excel_file_path)
        counter = 1
        if os.path.getsize(csv_file_path) > 0 :                                
            
            with pd.ExcelWriter(excel_file_path, engine='xlsxwriter') as writer:
                for i, chunk in enumerate(pd.read_csv(csv_file_path, chunksize=chunk_size, keep_default_na=False, na_filter=False, dtype=str)):
                    start_time = timeit.default_timer()                    
                    chunk.to_excel(writer, sheet_name='Sheet', index=False, startrow=i * chunk_size, header=False)
                    elapsed_time = timeit.default_timer() - start_time
                    print(f'writing chunk completed in {elapsed_time} seconds for chunk number : {counter} ')
                    counter += 1
                print(f'Successfully converted file : {csv_file_path}')
    except Exception as e:
        print('Error encountered in process_csv_files ' + str(e))

try:
    start_time_main = timeit.default_timer()
    output_base_path = 'output_directory_path'
    csv_file_path = '/input_directory_path/example.csv'
    chunk_size = 10000      
    process_csv_files(csv_file_path, output_base_path, chunk_size)
    elapsed_time = timeit.default_timer() - start_time_main
    print(f'Successfully converted the job in : {elapsed_time} seconds.')     
    
except Exception as e:
    print('Error encountered in csv_to_excel_convertor ' + str(e))

已尝试的优化措施 #

• 调整最优分块大小（10000为最优值）
• 更换XLSX写入引擎（xlsxwriter表现最佳）
• 指定合适的数据类型（仅降低内存占用，写入耗时未改善）

当前单个分块（10000行×177列）处理耗时约13秒，求进一步提速方法。

提速方案建议 #

1. 启用xlsxwriter的constant_memory模式 #

xlsxwriter默认会为单元格添加默认格式，且缓存工作表数据，这会显著增加写入耗时。启用constant_memory模式可以跳过默认格式、逐行写入，大幅降低时间开销：

with pd.ExcelWriter(excel_file_path, engine='xlsxwriter', 
                    engine_kwargs={'options': {'constant_memory': True}}) as writer:

2. 绕过Pandas，直接用xlsxwriter写入数据 #

Pandas的to_excel封装会带来额外开销，直接将分块数据转为列表后调用xlsxwriter底层接口写入：

with pd.ExcelWriter(excel_file_path, engine='xlsxwriter') as writer:
    worksheet = writer.book.add_worksheet('Sheet')
    # 先写入表头
    header = pd.read_csv(csv_file_path, nrows=0).columns.tolist()
    worksheet.write_row(0, 0, header)
    start_row = 1
    counter = 1
    for chunk in pd.read_csv(csv_file_path, chunksize=chunk_size, keep_default_na=False, na_filter=False, dtype=str):
        start_time = timeit.default_timer()
        # 将分块转为二维列表
        data = chunk.values.tolist()
        # 批量写入行
        for row_num, row_data in enumerate(data, start=start_row):
            worksheet.write_row(row_num, 0, row_data)
        start_row += chunk_size
        elapsed_time = timeit.default_timer() - start_time
        print(f'写入分块 {counter} 耗时 {elapsed_time} 秒')
        counter += 1

3. 用原生csv模块读取，彻底跳过Pandas转换 #

如果不需要Pandas的数据处理能力，直接用csv模块读取文件，生成列表后写入xlsxwriter，能最大程度减少内存和时间开销：

import csv

with pd.ExcelWriter(excel_file_path, engine='xlsxwriter') as writer:
    worksheet = writer.book.add_worksheet('Sheet')
    with open(csv_file_path, 'r', encoding='utf-8') as f:
        reader = csv.reader(f)
        # 写入表头
        header = next(reader)
        worksheet.write_row(0, 0, header)
        start_row = 1
        counter = 1
        chunk = []
        for row in reader:
            chunk.append(row)
            if len(chunk) == chunk_size:
                start_time = timeit.default_timer()
                for idx, data_row in enumerate(chunk, start=start_row):
                    worksheet.write_row(idx, 0, data_row)
                elapsed_time = timeit.default_timer() - start_time
                print(f'写入分块 {counter} 耗时 {elapsed_time} 秒')
                counter += 1
                chunk = []
                start_row += chunk_size
        # 处理剩余行
        if chunk:
            for idx, data_row in enumerate(chunk, start=start_row):
                worksheet.write_row(idx, 0, data_row)

4. 多进程+临时文件合并（超大文件场景） #

若CPU核心充足，可将分块写入拆分为多进程（注意xlsxwriter不支持多进程写同一个文件），先写入临时XLSX文件，最后用openpyxl合并：

from multiprocessing import Pool
import openpyxl

def process_chunk(chunk_data):
    chunk, temp_file, start_row = chunk_data
    with pd.ExcelWriter(temp_file, engine='xlsxwriter', engine_kwargs={'options': {'constant_memory': True}}) as writer:
        chunk.to_excel(writer, sheet_name='Sheet', index=False, startrow=start_row, header=False)
    return temp_file

# 主流程
temp_files = []
chunk_list = []
header = pd.read_csv(csv_file_path, nrows=0).columns.tolist()

for i, chunk in enumerate(pd.read_csv(csv_file_path, chunksize=chunk_size, keep_default_na=False, na_filter=False, dtype=str)):
    temp_file = f'temp_chunk_{i}.xlsx'
    chunk_list.append((chunk, temp_file, i*chunk_size))
    temp_files.append(temp_file)

# 多进程处理分块
with Pool(processes=os.cpu_count()) as pool:
    pool.map(process_chunk, chunk_list)

# 合并临时文件到最终文件
final_wb = openpyxl.Workbook()
final_ws = final_wb.active
final_ws.title = 'Sheet'
final_ws.append(header)

for temp_file in temp_files:
    wb = openpyxl.load_workbook(temp_file)
    ws = wb['Sheet']
    for row in ws.iter_rows(min_row=2, values_only=True):
        final_ws.append(row)
    wb.close()
    os.remove(temp_file)

final_wb.save(excel_file_path)

内容的提问来源于stack exchange，提问作者m_beta