☰
大CSV文件拆分实战:Python脚本按行与条件切分指南
2026/10/16 8:22:50 网站建设 项目流程

简介:处理大规模数据时,CSV文件常因体积过大而无法直接导入MATLAB,这份免费绿色工具正是面向此类高频痛点的轻量解决方案,适合数据分析人员、科研用户及MATLAB初学者使用。压缩包内共3个文件,包含可执行程序、说明文档及下载来源快捷方式,整体大小仅10KB,无需安装、不写注册表,便携易用。目前已有1136人浏览学习,工具支持按行数或文件大小设定拆分条件,自动将大文件批量分割为多个小文件,配合MATLAB的readtable函数循环读取,可显著降低内存占用、避免溢出问题;说明文档还给出了详细操作指引与注意事项,即使初次接触也能快速掌握。除MATLAB外,该工具同样适用于Excel、Python等常用数据分析环境,为处理超大CSV文件提供了一条直接、高效且免手动干预的路径。 先说个很实际的场景:运营发来一个几GB的“手机价格预测.csv”,让我帮忙核对里面某几列的数据。我电脑上的Excel直接卡死,双击后转了五分钟最后还是“未响应”;换数据库导入,又因为字段里有肉眼看不到的换行符导致解析错位。这时候第一时间想到的就是做一个够轻量的 csv拆分工具,把大文件按行数切成小份,既能用Excel直接打开,又能分批次喂给下游的统计脚本。

我相信不少人都被“大CSV”折磨过。这个文件格式看起来就是“逗号分隔的文本”,可真要动手拆,里面藏着编码、分隔符、引号换行、内存占用一堆问题。这篇文章我就把自己手搓拆分脚本的完整思路、代码和踩坑经验整理出来,给同样在处理CSV的朋友做一个可以直接抄作业的参考。

1. 拆CSV到底卡在哪:先分清你属于哪种“拆”

1.1 三个真实场景,需求各不相同

场景一:文件太大,Excel打不开,想看个大致内容都费劲。这种拆法最朴素,按“每个输出文件多少行”来切,比如50万行一个小文件。切完以后,每个小文件用Excel、WPS或VS Code打开都不吃力。

场景二:某个特定分类的数据要单独发给别人。比如“电力负荷数据气象csv”里有一列是地区编码,你想把华东、华北、华南的数据分别拆出来;或者“手机价格预测.csv”里想按品牌、按价格区间拆。这种拆法不是按行数,而是按条件过滤,常见的有两类:按某一列的值拆分到一个文件,按某一个布尔条件拆到两个文件。

场景三:文件需要分批导入数据库,但DB引擎对单次导入的行数有上限。这时拆出来的文件最好还保留原始表头,每份文件都是一张结构完整的小表,导入时不会因为缺列名而报错。

这三种需求本质上都叫“拆CSV”,但实现方式不同。我自己写脚本时会把它们拆成两个函数:一个按行数均分,一个按条件分组,避免互相纠缠。

1.2 CSV这个格式,比你以为的复杂

CSV不是像Excel那种封闭的二进制格式,它只是一个约定俗成的文本规范:一列用分隔符隔开,一行用换行符结束。问题就出在“约定俗成”这四个字上。

  • 分隔符可能是逗号,也可能是分号、Tab,甚至竖线|。
  • 编码可能是UTF-8、带BOM的UTF-8、GBK、GB18030,甚至Latin-1。
  • 更麻烦的是,一个字段如果包含逗号、换行,必须用双引号包起来,例如"北京, 上海", 2024, 100。这时如果你直接按row.split(',')去切,或者按\n去分行,数据从中间就断开了。

所以,真正稳妥的拆分方式,是用语言自带的CSV解析模块去逐行读取,而不是自己写字符串切分的逻辑。Python的csv标准库就是专门处理引号、转义和换行嵌套的,这也是我后面选Python的主要原因。

2. 工具选型:现成软件、命令行、还是自己写脚本

2.1 现成工具的适用边界

市面上不是没有CSV拆分工具,网页版在线拆分、桌面版CSV Sorter、Excel里的Power Query,都能做一部分。它们的优点是无脑上手,但天花板也很明显:

  • 在线工具要把整个文件上传到服务器,几GB的文件传半天,而且数据隐私也让人不放心。
  • 桌面CSV工具通常能处理几百MB,但遇到超大文件或带嵌套换行的脏数据,一样会翻车。
  • Excel和Power Query适合“小批量手动操作”,但它的行数上限和内存模型决定了,处理百万行级别的数据已经很吃力,更不用说拆分后还要继续处理几十个文件。
  • 命令行里有个split -l 100000命令,能按行数切文本,但它不管表头,也不管引号里的换行符,拆出来的数据大概率是残缺的。

所以如果是自己电脑上的临时任务,用现成工具没问题;但如果这个“拆CSV”会反复出现,或者要纳入日常数据处理流程,我建议直接把拆分逻辑固化成脚本,一劳永逸。

2.2 为什么最终选了Python,而不是C#或纯命令行

热搜词里有“c# 数据保存到csv表格”,说明不少C#开发者也经常在处理CSV。C#当然能做,写个CsvHelper的读取循环也不难,但做一个“一次性工具脚本”时,它的开发成本偏高:要建工程、引用NuGet包、处理类型声明,折腾完可能半小时就过去了。

Python的优势在于标准库自带csv,不用装任何第三方包就能解析和生成CSV;而且它是解释型语言,拿到任何一个环境,改改路径就能跑。和大数据生态又是天然衔接的,拆完之后如果还要用pandas做分析,脚本直接改就能合并到流程里。

有人可能会说,pandas本身不就能拆分吗?pd.read_csv读完,df.groupby、df.iloc随便拆。但在文件特别大的场景下,pandas一次性读入内存的代价非常大,很可能读取阶段就直接内存溢出了。真正稳妥的做法是用csv.reader流式读取,逐行判断、逐行写入,内存占用几乎可以忽略。这也是我在下面的脚本里坚持不用pandas的原因。

3. 一个能直接抄走的拆分脚本:按行拆且保留表头

3.1 命名规则与参数设计

先明确输出文件的命名规则。假设源文件叫手机价格预测.csv,按每20万行拆分,输出目录split_output下的文件命名如下:

  • 手机价格预测_part1.csv
  • 手机价格预测_part2.csv
  • 依次递增

每个输出文件都包含原始表头,这样后续每个小文件都能独立打开、独立入库。“每个文件保留表头”这一点非常关键,很多拆分行数的工具会把表头丢掉,数据就废了一半。

3.2 完整代码与关键点注释

先看一个可以直接运行的最小版本:

import csv import os import argparse def split_csv_by_rows(input_file, rows_per_file=200000, out_dir="split_output", has_header=True): """ 按行数拆分CSV,每个输出文件都保留表头。 rows_per_file: 每个文件包含的数据行数(不含表头) """ os.makedirs(out_dir, exist_ok=True) base_name = os.path.splitext(os.path.basename(input_file))[0] header = None current_lines = 0 part_no = 1 output_fp = None writer = None def open_next_output(): nonlocal output_fp, writer, current_lines if output_fp is not None: output_fp.close() out_path = os.path.join(out_dir, f"{base_name}_part{part_no}.csv") output_fp = open(out_path, "w", encoding="utf-8", newline="") writer = csv.writer(output_fp) if header is not None: writer.writerow(header) current_lines = 0 print(f"正在写入: {out_path}") with open(input_file, "r", encoding="utf-8", newline="") as f: reader = csv.reader(f) if has_header: header = next(reader) open_next_output() for row in reader: if current_lines >= rows_per_file: part_no += 1 open_next_output() writer.writerow(row) current_lines += 1 if output_fp is not None: output_fp.close() print(f"拆分完成,共生成 {part_no} 个文件") if __name__ == "__main__": parser = argparse.ArgumentParser(description="CSV按行数拆分工具") parser.add_argument("input_file", help="输入CSV文件路径") parser.add_argument("-r", "--rows", type=int, default=200000, help="每个文件的数据行数") parser.add_argument("-o", "--out_dir", default="split_output", help="输出目录") parser.add_argument("--no-header", action="store_true", help="如果CSV没有表头,加这个参数") args = parser.parse_args() split_csv_by_rows( args.input_file, rows_per_file=args.rows, out_dir=args.out_dir, has_header=not args.no_header, )

几个容易被忽略的细节:

newline=""必须加。打开输入和输出文件时如果不指定newline="",csv模块在Windows上会把\r\n再转换成一次,导致输出文件的行尾变成\r\r\n,Excel打开时会出现空行。

流式写入,不要攒批。这段代码是读一行、写一行,内存占用几乎恒定。不管源文件是200MB还是2GB,脚本都不会因为“一次性读入”而爆内存。有些人的写法是读一批攒在列表里再统一写,如果一批是50万行,内存照样压力很大,没必要。

part_no从一开始编号。这是因为逻辑上先创建了第一个文件,再在后续循环中判断是否需要切换。输入文件为空时也能正确处理,不会生成空文件。

3.3 如果CSV没有表头,或者有多行表头怎么办

有的系统导出的CSV前两行才是表头,比如第一行是生成时间,第二行才是字段名。处理方法是,在读入时先把前面N行单独取出,每个输出文件写入时原样复制这N行,再把数据行的写入逻辑接在后面。

with open(input_file, "r", encoding="utf-8", newline="") as f: reader = csv.reader(f) header_lines = [next(reader) for _ in range(2)] # 读取前两行作为表头 ...

我在实际项目里遇到过导出文件首行是“报表生成时间:2024-05-01”这种无效行的情况,这种情况下如果不做处理,表头就错位了。

4. 真正会咬人的坑:我替你们踩过的问题清单

4.1 中文乱码与UTF-8 BOM

很多Windows环境下生成的CSV其实是GBK编码,而Python默认按UTF-8去读,读出来的中文会直接乱码,甚至引发UnicodeDecodeError。解决方式是在打开文件时指定编码:

open(input_file, "r", encoding="gbk", newline="")

更稳妥的做法是“自动探测”:先用二进制模式读一小段,尝试用UTF-8解码,失败再换GBK。网上有chardet库可以检测编码,但为了一个脚本去装第三方库不值得,我自己通常写一个简单的尝试解码函数:

def detect_encoding(file_path): with open(file_path, "rb") as f: raw = f.read(4096) if raw.startswith(b"\xef\xbb\xbf"): return "utf-8-sig" try: raw.decode("utf-8") return "utf-8" except UnicodeDecodeError: return "gbk"

这里有个细节:如果是UTF-8且带BOM,读取时用utf-8-sig,这样Python会剥掉BOM;但写入新文件时如果不想每个文件都带BOM,统一改成utf-8即可。如果你拆出来的文件还要用Excel打开,Excel对UTF-8无BOM的识别偶尔会抽风,那时候就在输出时也写成utf-8-sig,虽然每个文件多了三个字节,但兼容性最好。

4.2 引号字段里的逗号和新行

这是我见过最多人翻车的地方。CSV里像"北京市, 朝阳区", 100000这样的行,直接用line.split(",")去切,会莫名其妙多出一列;如果字段内容里有换行,比如备注列填了多行文字,直接用for line in file去读,更是会把一条记录拆成两行。

所以我的建议很简单:处理CSV一律用语言自带的 csv 模块,别自己拼split逻辑。csv模块会处理引号内的逗号,也会把引号内的换行视为字段的一部分。代价是它读起来会比单纯按行split慢一点,但换来的是正确性,值得。

4.3 千万不要一次性读进内存

之前一个朋友用open(...).readlines()把700MB的CSV全读进内存,然后拆分,结果连操作系统都开始卡。原因很简单,700MB的文件按行拆成Python字符串列表,内存占用可能到3~5GB。所以无论文件多大,只要没有特别需求,就用迭代器逐行读,不要readlines(),不要pandas.read_csv()。写脚本时优先考虑流式处理,这是大文件处理的第一原则。

4.4 拆分结果再入库时,注意SQL Server等数据库的格式要求

热搜词里有“sql server 的导入csv”,这其实是个常见衔接场景:我们把大CSV拆成小文件后,需要批量导入SQL Server。这时要留意SQL Server的BULK INSERT或bcp工具对CSV的格式预期:默认编码、字段终止符、行终止符可能都不太一样。拆出来的文件如果带着BOM,有时会让第一列列名多一个不可见字符,导入时报“列名无效”。

遇到这种情况,最省事的方式是把输出编码固定为utf-8(不带BOM),并在SQL导入时显式指定CODEPAGE='65001'。如果下游系统只认GBK,那就统一输出gbk。总之,“输出编码”要和下游系统提前对齐,别等到导入报错再返工。

5. 进阶玩法:按条件拆、自动识别分隔符、编码自适应

5.1 按某一列的值分组拆分

假设“手机价格预测.csv”里有一列是brand,你想让每个品牌单独一个文件,循环里判断就够:

import csv import os from collections import defaultdict def split_csv_by_column(input_file, column_name="brand", out_dir="split_by_brand", has_header=True): os.makedirs(out_dir, exist_ok=True) base_name = os.path.splitext(os.path.basename(input_file))[0] header = None writers = defaultdict(dict) # value -> {"fp": output_fp, "writer": csv.writer} with open(input_file, "r", encoding="utf-8", newline="") as f: reader = csv.reader(f) if has_header: header = next(reader) col_idx = header.index(column_name) else: col_idx = 0 for row in reader: if col_idx >= len(row): continue val = row[col_idx] if not val: val = "empty" if val not in writers: out_path = os.path.join(out_dir, f"{base_name}_{val}.csv") fp = open(out_path, "w", encoding="utf-8", newline="") wr = csv.writer(fp) if header is not None: wr.writerow(header) writers[val] = {"fp": fp, "writer": wr} writers[val]["writer"].writerow(row) for info in writers.values(): info["fp"].close() print(f"按列 {column_name} 拆分完成,共 {len(writers)} 个分组")

这个版本有个隐患:如果按某一列拆分后分组个数特别多,比如几千个品牌,就会同时打开几千个文件句柄,操作系统容易报 “Too many open files”。我实际使用时通常先做一次分组统计,确认大概有几个分组,或者改成“边读边写,但每写满N行就关闭重新打开”,避免句柄耗尽。

5.2 自动识别分隔符

很多CSV的“C”其实是逗号,但总有一些文件用了分号、Tab,甚至竖线。自动识别分隔符也不复杂,读头几行,统计各类候选分隔符在引号外的出现次数,出现次数最多的那个大概率就是分隔符。其实Python的csv模块在csv.Sniffer里已经实现了这个能力:

with open(input_file, "r", encoding=detect_encoding(input_file), newline="") as f: sample = f.read(8192) dialect = csv.Sniffer().sniff(sample, delimiters=",;\t|") print("识别到的分隔符:", dialect.delimiter)

注意Sniffer偶尔会误判,尤其是在数据只有一两列、或某列内部大量使用逗号时。稳妥做法是识别完之后,人工打印确认一下,不要直接闷头处理整个大文件。

5.3 输出完做一次完整性校验

拆完文件后,我最怕的是交接给下游时才发现数据少了几行。所以在系统进去“自动化定期执行”之前,我会在脚本最后加一段校验逻辑:统计源文件的总数据行数,再统计所有输出文件的行数总和,两边一比对,对不上就报警。这个校验的成本很低,但能省掉大量排查时间。

total_out_rows = 0 for fn in os.listdir(out_dir): with open(os.path.join(out_dir, fn), "r", encoding="utf-8", newline="") as f: reader = csv.reader(f) next(reader, None) # 跳过表头 total_out_rows += sum(1 for _ in reader) print(f"源文件数据行数: {source_rows}, 输出文件数据行数合计: {total_out_rows}")

如果source_rows != total_out_rows,问题多半出在脏数据行上,比如某行字段数不一致、引号不闭合,csv模块可能把它当成空行跳过或解析异常。遇到这种情况,我会先用文本编辑器打开CSV原始文件,检查那几行的手工数据来源,而不是盲目重跑。

6. 关于“拆分工具还能怎么扩展”的一点个人体会

这些年我用CSV拆分工具处理过不少实际数据,包括“电力负荷数据气象csv”“手机价格预测.csv”,也配合过“sql server 的导入csv”“word文档里怎么批量插入csv文档附件”这些场景。每次用得越顺手,越觉得这种“小工具”不该只是临时写一遍就扔,建议把按行拆、按列拆、编码探测、分隔符识别这几个函数收进自己的公共工具库。下次拿到一个新CSV,先让脚本自动识别编码和分隔符,再决定用哪种拆分策略,整个过程不超过十秒。

最后分享一个小技巧:拆分大CSV时,源文件所在的磁盘最好和输出目录不在同一块物理盘上。因为读写同时进行时,如果同盘IO,拆一个2GB的文件可能要等很久;分盘后速度提升非常明显。另外,拆完后不要急着删源文件,先抽查输出文件的表头、行数和随机几行内容再清理,这是我的习惯,也建议你保留这个习惯。

本文还有配套的精品资源,点击获取

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询