简介这是一款面向数据分析师、大数据工程师及IT从业者的CSV大文件分割工具专门解决百万行级CSV文件难以用普通编辑器或表格软件打开、处理的痛点。工具支持按行数、文件大小或指定列值等条件将大文件快速拆分为多个小文件便于后续导入、备份、传输与分析并可配合Hadoop生态下的MapReduce、Hive等组件完成数据预处理。资源包共4个文件以exe可执行程序为主辅以txt使用说明、htm下载说明及url帮助链接整体约533KB解压后即可运行无需复杂配置。目前已有5567人学习下载适合需要处理大规模CSV数据、优化数据导入流程的读者参考使用可帮助快速完成数据分割与预处理提升大数据场景下的处理效率。1. 几百万行的 csv 打不开问题从来不在 Excel你手上有个 2GB 的 csv双击打开Excel 转圈转到天荒地老最后弹一句「文件过大无法完全加载」。或者你从数据库导出一份手机信令明细几千万行想按地市拆开分给不同同事处理手动复制粘贴根本不可能。这时候你需要的不是更贵的电脑而是一个 csv 大文件分割工具——把单个巨型 csv 按行数、按字段值、按文件体积切成若干个小文件让下游的 pandas、MySQL、WPS 都能吃得下。这个方向值不值得做非常值。csv 是数据交换里最土也最通用的格式导入 csv 文件、pandas 读取 csv 文件、mysql 导入 csv 文件这些操作每天都在发生而「文件太大」是卡住整条链路的第一道坎。本文面向三类人手里有超大 csv 打不开的分析师、要把 csv 灌进数据库的工程师、以及想自己写一个分割脚本的开发者。我会把分割的几种策略、参数怎么定、Python 和命令行两条落地路径、以及我踩过的坑一次讲清楚照着做就能复现。2. csv 分割的三种切法按行、按字段、按体积2.1 先想清楚你要哪种「小」很多人一上来就问「怎么分割」但真正该先回答的是「按什么维度分」。csv 分割不是只有一种切法选错了后面全是返工。按行数切是最常见的需求。比如一份 500 万行的订单明细你想每 50 万行一个文件切出 10 个。这种切法适合后续要并行处理、或者要分批导入数据库的场景。它的优点是实现简单、每个文件大小均匀缺点是如果数据本身有分组结构比如同一个用户的多条记录按行切会把同一组数据劈到两个文件里。按字段值切是业务导向的切法。比如你有一份 2023 年全国区县级手机信令数据 csv想按省份拆成 31 个文件每个省一个。这种切法保留了业务分组下游按省处理时不用再过滤。实现上需要先扫一遍目标列、收集所有唯一值再决定开几个输出文件句柄。按文件体积切适合「我不管多少行我只要每个文件不超过 100MB」的场景比如要往有上传大小限制的系统里传。这种切法要边写边统计字节数写满一个就换下一个实现上比按行数稍麻烦因为不能提前算好切点。提示三种切法可以组合。我一般会先按字段值分组再在组内按行数二次切分这样既保留业务结构又避免某个省的数据特别大导致单文件过大。2.2 流式读取为什么是唯一正确姿势不管哪种切法核心原则只有一条绝对不要把整个文件读进内存。pandas.read_csv默认是全量加载一个 2GB 的 csv 读进来内存占用轻松到 610GB机器直接爆掉。这不是 pandas 的错是用法不对。正确做法是流式读取一行一行或者一块一块地处理处理完就丢。Python 里最稳的是标准库csv模块配合文件迭代它不会把整个文件载入内存。如果一定要用 pandas就用chunksize参数它返回一个迭代器每次只给你一块 DataFrame。import csv # 流式读取每次只处理一行内存占用恒定 with open(big.csv, r, encodingutf-8, newline) as f: reader csv.reader(f) header next(reader) # 先取出表头后面每个分片都要写回去 for row in reader: # 这里处理单行内存里永远只有一行 pass这段代码的关键在newline。csv 模块官方文档明确要求打开文件时传这个参数否则在 Windows 上遇到字段内嵌换行符时会解析错乱。encoding要按实际文件来国内数据常见utf-8和gbk两种选错了第一行就报UnicodeDecodeError。next(reader)单独取表头是个习惯动作因为分割后的每个小文件都得带上表头不然下游读取时列名就丢了。2.3 按行数分割的最小可用实现先给一个能直接跑的按行数分割脚本这是最常用的版本。import csv import os def split_by_rows(src_path, out_dir, rows_per_file, encodingutf-8): 按固定行数分割 csv src_path: 源文件路径 out_dir: 输出目录 rows_per_file: 每个分片的最大数据行数不含表头 os.makedirs(out_dir, exist_okTrue) base os.path.splitext(os.path.basename(src_path))[0] with open(src_path, r, encodingencoding, newline) as f: reader csv.reader(f) header next(reader) file_idx 1 row_count 0 writer None out_f None def open_new(): nonlocal writer, out_f, row_count if out_f: out_f.close() name f{base}_part{file_idx:03d}.csv out_f open(os.path.join(out_dir, name), w, encodingencoding, newline) writer csv.writer(out_f) writer.writerow(header) # 每个分片都写表头 row_count 0 open_new() for row in reader: if row_count rows_per_file: file_idx 1 open_new() writer.writerow(row) row_count 1 if out_f: out_f.close() split_by_rows(big.csv, output, 500000)逻辑上open_new()负责关旧文件、开新文件、写表头、重置计数。主循环每读一行就判断是否写满写满就换文件。{file_idx:03d}保证文件名按part001、part002排序不会出现part10排在part2前面的问题。参数上rows_per_file怎么定没有标准答案看下游。要导入 MySQL单文件 50 万到 100 万行比较舒服LOAD DATA一次跑完不超时。要给同事用 WPS 打开看那得压到 10 万行以内不然 WPS 一样卡。encoding必须和源文件一致不确定就先用chardet探一下或者拿记事本另存为时看编码选项。3. 按字段值分割手机信令、订单、日志的分组拆解3.1 两遍扫描法先收集唯一值再分流按字段值分割比按行数复杂一点因为你要先知道有哪些分组。最稳的做法是两遍扫描第一遍只读目标列收集所有唯一值第二遍才真正分流写入。import csv from collections import defaultdict def split_by_column(src_path, out_dir, col_name, encodingutf-8): 按指定列的值分割每个唯一值一个文件 # 第一遍找出目标列的索引和所有唯一值 with open(src_path, r, encodingencoding, newline) as f: reader csv.reader(f) header next(reader) col_idx header.index(col_name) unique_vals set() for row in reader: if len(row) col_idx: unique_vals.add(row[col_idx]) print(f共发现 {len(unique_vals)} 个分组) # 第二遍分流写入 import os os.makedirs(out_dir, exist_okTrue) handles {} writers {} for val in unique_vals: safe str(val).replace(/, _).replace(\\, _) path os.path.join(out_dir, f{safe}.csv) fh open(path, w, encodingencoding, newline) w csv.writer(fh) w.writerow(header) handles[val] fh writers[val] w with open(src_path, r, encodingencoding, newline) as f: reader csv.reader(f) next(reader) for row in reader: if len(row) col_idx: val row[col_idx] if val in writers: writers[val].writerow(row) for fh in handles.values(): fh.close()第一遍扫描只读一列内存里存的是唯一值集合通常远小于总行数。第二遍为每个唯一值开一个文件句柄边读边写。safe那行是防止字段值里有斜杠导致路径出错比如日期字段2023/01/01直接当文件名会创建子目录。这里有个硬限制同时打开的文件句柄数。Linux 默认单进程 1024 个如果唯一值有几千个open会直接抛Too many open files。解决办法是分批处理或者用defaultdict缓存行、写满一批再落盘。手机信令按区县分可能有 2800 多个分组就必须分批。3.2 分组数太多时的分批落盘策略当唯一值超过几百个就不能一次性全开句柄了。我一般用「缓冲 定期 flush」的方式维护一个字典key 是分组值value 是待写入的行列表攒够一定数量就统一追加写入。import csv, os from collections import defaultdict def split_by_column_batched(src_path, out_dir, col_name, batch_size5000, encodingutf-8): os.makedirs(out_dir, exist_okTrue) buffers defaultdict(list) header None col_idx None def flush(): for val, rows in buffers.items(): if not rows: continue safe str(val).replace(/, _).replace(\\, _) path os.path.join(out_dir, f{safe}.csv) exists os.path.exists(path) with open(path, a, encodingencoding, newline) as fh: w csv.writer(fh) if not exists: w.writerow(header) w.writerows(rows) buffers.clear() with open(src_path, r, encodingencoding, newline) as f: reader csv.reader(f) header next(reader) col_idx header.index(col_name) pending 0 for row in reader: if len(row) col_idx: buffers[row[col_idx]].append(row) pending 1 if pending batch_size: flush() pending 0 flush()batch_size控制内存里最多攒多少行5000 到 20000 都行看单行宽度。flush()用追加模式a打开第一次写表头之后只追加数据。这样无论有多少分组同时打开的文件句柄永远只有 1 个。代价是每个分组会被打开关闭多次磁盘 IO 比一次性写入多但在 SSD 上可以接受。3.3 参数怎么定编码、分隔符、引号csv 看着简单参数不对全是坑。下面这张表是我处理国内数据时总结的常用配置。参数常见取值什么时候用encodingutf-8 / gbk / utf-8-sig源文件带 BOM 用 utf-8-sig否则第一列名会多个不可见字符delimiter, / \t / ;欧洲数据常用分号日志导出常用制表符quotechar字段内含逗号时靠它包裹默认就是双引号newline打开文件时必传否则内嵌换行解析错乱utf-8-sig这个坑我踩过不止一次。Windows 上用 Excel 另存为 csv默认会加 BOM 头用utf-8读出来第一列列名是\ufeffid而不是id后面header.index(id)直接报错。遇到这种文件读的时候用utf-8-sig它会自动吃掉 BOM。分隔符也要确认。有些系统导出的「csv」其实是制表符分隔的 tsv只是扩展名写成了 csv。用逗号去切整行会变成一个字段。判断方法很简单读第一行看有几个逗号如果一个都没有但有很多制表符那就是 tsv。4. 命令行方案不写代码也能切大文件4.1 split 命令一行搞定按行数切不是所有人都想写 Python。Linux 和 macOS 自带的split命令就能按行数切文件速度极快因为它是 C 实现的。# 每 50 万行一个文件后缀用数字保留 .csv 扩展名 split -l 500000 -d --additional-suffix.csv big.csv part_ # 结果part_00.csv, part_01.csv, ...-l 500000指定每个文件的行数-d用数字后缀而不是字母--additional-suffix.csv给每个分片加扩展名。注意split是按物理行切的如果 csv 字段里有内嵌换行它会把一行数据劈成两半这是它最大的局限。数据干净、没有内嵌换行时split是最快的选择。还有个细节split不会给每个分片加表头。如果下游需要表头得手动补。# 先取出表头 head -n 1 big.csv header.csv # 给每个分片补表头 for f in part_*.csv; do cat header.csv $f $f.tmp mv $f.tmp $f done这个循环对每个文件做一次「表头 内容」的拼接文件多的时候有点慢但胜在简单。注意mv那步直接cat header.csv $f $f会把文件清空必须先写临时文件再改名。4.2 csvkit 与 miller带表头感知的命令行工具split不懂 csv 语义csvkit和miller懂。csvkit是一套 Python 写的 csv 命令行工具集miller是 Go 写的性能更好。# 用 miller 按字段值分割每个城市一个文件 mlr --csv split -g city big.csv # 结果生成 city北京.csv、city上海.csv 等mlr --csv split -g city里的-g指定分组字段它会自动处理表头、引号、编码。这是我最推荐的命令行方案尤其是数据有内嵌逗号或换行时split会翻车miller不会。csvkit的对应工具是csvsplit用法类似# 按列的值分割 csvsplit -c 3 big.csv-c 3表示按第 3 列的值分组。csvkit基于 Python速度比miller慢但如果你已经在 Python 生态里装一个很方便。注意命令行工具默认按 UTF-8 处理遇到 GBK 编码的中文 csv 可能乱码。miller可以用--csv --from file --ifs ,配合编码参数但更省事的办法是先用iconv转成 UTF-8 再处理。5. 避坑与排查分割脚本翻车的五个现场5.1 分片第一行变成乱码或列名错位现象切完之后打开某个分片第一行不是表头而是一条数据或者表头里混进了数据。原因主循环里忘了在换文件时重新写表头或者表头被当成了数据行处理。常见于用next(reader)取表头后换文件时没有再次writer.writerow(header)。解决把「写表头」这个动作封装进开新文件的函数里每次open_new()都执行一次。不要依赖主循环里手动补。5.2 内存不降反升跑一半被 OOM kill现象明明用了流式读取内存还是越跑越高最后被系统杀掉。原因多半是用了pandas.read_csv没加chunksize或者把每行都 append 进了一个列表。还有一种隐蔽情况csv.reader本身是流式的但你把list(reader)转成了列表等于全量加载。解决检查代码里有没有list(reader)、readlines()、read().split()这类全量操作。用pandas时确认chunksize参数存在并且是在for chunk in pd.read_csv(..., chunksize100000)里迭代。5.3 字段里的逗号把一行切成两半现象分片里某些行的列数不对比表头少或者多。原因源文件字段里含逗号但没有用引号包裹或者用了非标准分隔符。csv.reader默认按逗号切遇到裸逗号就多切一刀。解决先确认源文件是不是标准 csv。用csv.Sniffer探测分隔符或者直接看第一行。如果字段里确实有逗号正确做法是源文件用双引号包裹字段读取时csv.reader会自动处理。如果源文件本身就不规范那得先清洗。5.4 中文文件名或字段值导致路径报错现象按字段值分割时某些分组直接报FileNotFoundError或创建了奇怪的目录。原因字段值里含/、\、:、*等文件系统保留字符。比如日期2023/01/01会被当成三级目录。解决写文件名前做一次替换把保留字符换成下划线。我一般用re.sub(r[\\/:*?|], _, str(val))统一处理。另外字段值过长也要截断文件名有 255 字节限制。5.5 切完发现行数对不上少了或多了现象所有分片行数加起来和源文件对不上。原因三种可能。一是源文件最后一行没有换行符某些工具会漏掉二是字段内嵌换行导致物理行数和逻辑行数不一致三是写入时缓冲区没 flush程序异常退出丢了数据。解决切完后做一次校验把所有分片的数据行数不含表头加起来和源文件总行数减一对比。用wc -l快速数物理行用csv.reader数逻辑行。写入端确保每个文件句柄都close()或者用with语句管理。6. 进阶把分割做成可复用的校验闭环分割脚本写完不是终点能验证结果对不对才是。我现在的习惯是任何分割任务都配一个校验脚本切完立刻跑一遍行数、列数、表头三项对齐才放心。import csv, os def verify_split(src_path, out_dir, encodingutf-8): 校验分割结果总行数、列数、表头一致性 with open(src_path, r, encodingencoding, newline) as f: src_rows sum(1 for _ in csv.reader(f)) - 1 # 减去表头 f.seek(0) src_header next(csv.reader(f)) total 0 for name in sorted(os.listdir(out_dir)): if not name.endswith(.csv): continue path os.path.join(out_dir, name) with open(path, r, encodingencoding, newline) as f: reader csv.reader(f) header next(reader) if header ! src_header: print(f[表头不一致] {name}) rows sum(1 for _ in reader) total rows print(f{name}: {rows} 行) print(f源文件 {src_rows} 行分片合计 {total} 行 f{一致 if src_rows total else 不一致}) verify_split(big.csv, output)这个校验脚本做三件事数源文件逻辑行数、逐个分片比对表头、累加分片行数。表头不一致通常意味着某个分片写表头时编码出了问题行数不一致则要回去查是不是有行被漏写或重复写。再进阶一点可以把分割和校验串成一条流水线用argparse包成命令行工具加上--by-rows、--by-column、--verify几个开关。这样团队里其他人不用看代码直接python splitter.py big.csv --by-rows 500000 --verify就能用。我自己的血泪经验是分割脚本最怕的不是慢是「看起来切完了但数据悄悄少了」。有一次按字段分割某个分组的值里带了个换行符csv.writer写出去之后下游用wc -l数行数对不上排查了半天才发现是字段内嵌换行。从那以后我养成了两个习惯一是切完必跑校验二是源文件先做一次csv模块的 round-trip 解析确认能被标准解析器正确读出再动手。数据这行没有后悔药多花五分钟校验省的是后面几小时的扯皮。希望帮到你。本文还有配套的精品资源点击获取