import pandas as pd import os from tqdm import tqdm import openpyxl def split_csv_to_excel(csv_file_path, output_excel_pathNone, rows_per_sheet5000, chunk_size10000): 将大型 CSV 文件按指定行数拆分到 Excel 的不同工作表中 优化版本适用于处理数百万行的超大数据集 参数: csv_file_path: 输入 CSV 文件路径 output_excel_path: 输出 Excel 文件路径默认为 CSV 同名文件 rows_per_sheet: 每个工作表包含的行数默认为 5000 chunk_size: 每次读取的 CSV 行数用于处理大文件 # 确定输出文件路径如果未指定输出路径则默认在原CSV文件名后加_split.xlsx if output_excel_path is None: base_name os.path.splitext(csv_file_path)[0] # 提取CSV文件的基础名称不含扩展名 output_excel_path f{base_name}_split.xlsx # 构建输出Excel文件名 # 获取 CSV 文件的列名表头读取第一行并按逗号分割 with open(csv_file_path, r, encodingutf-8) as f: # 以只读模式打开CSV文件指定编码为utf-8 header f.readline().strip().split(,) # 读取第一行去除首尾空格后按逗号分割为列表 # 计算 CSV 文件总行数用于进度显示遍历文件计数减去表头行 total_rows sum(1 for _ in open(csv_file_path, r, encodingutf-8)) - 1 # 生成器表达式遍历文件每一行,每一行对应生成一个1sum() 对所有的 1 求和计数再减1排除表头 # 计算需要读取的总块数总记录数除以每次读取的块大小向上取整 total_chunks (total_rows // chunk_size) (1 if total_rows % chunk_size ! 0 else 0) # 创建 Excel 工作簿并添加第一个工作表 wb openpyxl.Workbook() # 创建一个新的Excel工作簿 current_sheet 1 # 记录当前工作表编号从1开始 ws wb.active # 获取工作簿的活动工作表默认创建的第一个工作表 ws.title fSheet{current_sheet} # 为当前工作表命名如Sheet1 # 写入表头将CSV的列名写入Excel工作表的第一行 for col_idx, col_name in enumerate(header, 1): # 枚举列名索引从1开始Excel列索引从1开始 ws.cell(row1, columncol_idx, valuecol_name) # 在第1行、第col_idx列写入列名 current_row 2 # 当前工作表的写入行从第2行开始因为第1行是表头 total_sheets 1 # 记录总工作表数量初始为1 # 使用 tqdm 显示进度条按块读取CSV数据每次读取chunk_size行 # pd.read_csv(..., chunksizechunk_size)返回迭代器每次生成一个DataFrame数据块 for chunk in tqdm(pd.read_csv(csv_file_path, chunksizechunk_size), totaltotal_chunks, desc处理中): # 遍历数据块中的每一行_为索引row为行数据 for _, row in chunk.iterrows(): # 当达到工作表最大行数时创建新工作表 # 条件当前行号 单表最大行数 11是因为表头占1行 if current_row rows_per_sheet 1: ws wb.create_sheet(titlefSheet{current_sheet}) # 创建新工作表并命名 # 写入表头到新工作表 for col_idx, col_name in enumerate(header, 1): ws.cell(row1, columncol_idx, valuecol_name) current_row 2 # 重置当前行号为2新工作表从第2行开始写数据 total_sheets 1 # 总工作表数加1 # 将数据写入当前工作表遍历行中的每一列值 for col_idx, value in enumerate(row, 1): # 枚举行数据索引从1开始 ws.cell(rowcurrent_row, columncol_idx, valuevalue) # 在当前行、对应列写入值 current_row 1 # 当前行号加1准备写下一行数据 # 保存工作簿到指定路径 wb.save(output_excel_path) # 打印处理结果信息 print(fCSV 文件已成功拆分到 {output_excel_path}) print(f共生成 {total_sheets} 个工作表每个工作表包含最多 {rows_per_sheet} 条记录) # 使用示例当脚本直接运行时执行以下代码 if __name__ __main__: # 指定 CSV 文件路径 csv_path re_byticket_spider.csv # 调用函数进行拆分对于300万条数据建议增大chunk_size以提高效率 split_csv_to_excel(csv_path, rows_per_sheet5000, chunk_size50000)
将大型 CSV 文件按指定行数拆分到 Excel 的不同工作表中
发布时间：2026/5/26 19:15:59
import pandas as pd import os from tqdm import tqdm import openpyxl def split_csv_to_excel(csv_file_path, output_excel_pathNone, rows_per_sheet5000, chunk_size10000): 将大型 CSV 文件按指定行数拆分到 Excel 的不同工作表中 优化版本适用于处理数百万行的超大数据集 参数: csv_file_path: 输入 CSV 文件路径 output_excel_path: 输出 Excel 文件路径默认为 CSV 同名文件 rows_per_sheet: 每个工作表包含的行数默认为 5000 chunk_size: 每次读取的 CSV 行数用于处理大文件 # 确定输出文件路径如果未指定输出路径则默认在原CSV文件名后加_split.xlsx if output_excel_path is None: base_name os.path.splitext(csv_file_path)[0] # 提取CSV文件的基础名称不含扩展名 output_excel_path f{base_name}_split.xlsx # 构建输出Excel文件名 # 获取 CSV 文件的列名表头读取第一行并按逗号分割 with open(csv_file_path, r, encodingutf-8) as f: # 以只读模式打开CSV文件指定编码为utf-8 header f.readline().strip().split(,) # 读取第一行去除首尾空格后按逗号分割为列表 # 计算 CSV 文件总行数用于进度显示遍历文件计数减去表头行 total_rows sum(1 for _ in open(csv_file_path, r, encodingutf-8)) - 1 # 生成器表达式遍历文件每一行,每一行对应生成一个1sum() 对所有的 1 求和计数再减1排除表头 # 计算需要读取的总块数总记录数除以每次读取的块大小向上取整 total_chunks (total_rows // chunk_size) (1 if total_rows % chunk_size ! 0 else 0) # 创建 Excel 工作簿并添加第一个工作表 wb openpyxl.Workbook() # 创建一个新的Excel工作簿 current_sheet 1 # 记录当前工作表编号从1开始 ws wb.active # 获取工作簿的活动工作表默认创建的第一个工作表 ws.title fSheet{current_sheet} # 为当前工作表命名如Sheet1 # 写入表头将CSV的列名写入Excel工作表的第一行 for col_idx, col_name in enumerate(header, 1): # 枚举列名索引从1开始Excel列索引从1开始 ws.cell(row1, columncol_idx, valuecol_name) # 在第1行、第col_idx列写入列名 current_row 2 # 当前工作表的写入行从第2行开始因为第1行是表头 total_sheets 1 # 记录总工作表数量初始为1 # 使用 tqdm 显示进度条按块读取CSV数据每次读取chunk_size行 # pd.read_csv(..., chunksizechunk_size)返回迭代器每次生成一个DataFrame数据块 for chunk in tqdm(pd.read_csv(csv_file_path, chunksizechunk_size), totaltotal_chunks, desc处理中): # 遍历数据块中的每一行_为索引row为行数据 for _, row in chunk.iterrows(): # 当达到工作表最大行数时创建新工作表 # 条件当前行号 单表最大行数 11是因为表头占1行 if current_row rows_per_sheet 1: ws wb.create_sheet(titlefSheet{current_sheet}) # 创建新工作表并命名 # 写入表头到新工作表 for col_idx, col_name in enumerate(header, 1): ws.cell(row1, columncol_idx, valuecol_name) current_row 2 # 重置当前行号为2新工作表从第2行开始写数据 total_sheets 1 # 总工作表数加1 # 将数据写入当前工作表遍历行中的每一列值 for col_idx, value in enumerate(row, 1): # 枚举行数据索引从1开始 ws.cell(rowcurrent_row, columncol_idx, valuevalue) # 在当前行、对应列写入值 current_row 1 # 当前行号加1准备写下一行数据 # 保存工作簿到指定路径 wb.save(output_excel_path) # 打印处理结果信息 print(fCSV 文件已成功拆分到 {output_excel_path}) print(f共生成 {total_sheets} 个工作表每个工作表包含最多 {rows_per_sheet} 条记录) # 使用示例当脚本直接运行时执行以下代码 if __name__ __main__: # 指定 CSV 文件路径 csv_path re_byticket_spider.csv # 调用函数进行拆分对于300万条数据建议增大chunk_size以提高效率 split_csv_to_excel(csv_path, rows_per_sheet5000, chunk_size50000)
