ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

大CSV文件拆分实战:Python脚本按行与条件切分指南

大CSV文件拆分实战:Python脚本按行与条件切分指南 简介处理大规模数据时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_file200000, out_dirsplit_output, has_headerTrue): 按行数拆分CSV每个输出文件都保留表头。 rows_per_file: 每个文件包含的数据行数不含表头 os.makedirs(out_dir, exist_okTrue) 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, encodingutf-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, encodingutf-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(descriptionCSV按行数拆分工具) parser.add_argument(input_file, help输入CSV文件路径) parser.add_argument(-r, --rows, typeint, default200000, help每个文件的数据行数) parser.add_argument(-o, --out_dir, defaultsplit_output, help输出目录) parser.add_argument(--no-header, actionstore_true, help如果CSV没有表头加这个参数) args parser.parse_args() split_csv_by_rows( args.input_file, rows_per_fileargs.rows, out_dirargs.out_dir, has_headernot args.no_header, )几个容易被忽略的细节newline必须加。打开输入和输出文件时如果不指定newlinecsv模块在Windows上会把\r\n再转换成一次导致输出文件的行尾变成\r\r\nExcel打开时会出现空行。流式写入不要攒批。这段代码是读一行、写一行内存占用几乎恒定。不管源文件是200MB还是2GB脚本都不会因为“一次性读入”而爆内存。有些人的写法是读一批攒在列表里再统一写如果一批是50万行内存照样压力很大没必要。part_no从一开始编号。这是因为逻辑上先创建了第一个文件再在后续循环中判断是否需要切换。输入文件为空时也能正确处理不会生成空文件。3.3 如果CSV没有表头或者有多行表头怎么办有的系统导出的CSV前两行才是表头比如第一行是生成时间第二行才是字段名。处理方法是在读入时先把前面N行单独取出每个输出文件写入时原样复制这N行再把数据行的写入逻辑接在后面。with open(input_file, r, encodingutf-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, encodinggbk, 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导入时显式指定CODEPAGE65001。如果下游系统只认GBK那就统一输出gbk。总之“输出编码”要和下游系统提前对齐别等到导入报错再返工。5. 进阶玩法按条件拆、自动识别分隔符、编码自适应5.1 按某一列的值分组拆分假设“手机价格预测.csv”里有一列是brand你想让每个品牌单独一个文件循环里判断就够import csv import os from collections import defaultdict def split_csv_by_column(input_file, column_namebrand, out_dirsplit_by_brand, has_headerTrue): os.makedirs(out_dir, exist_okTrue) 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, encodingutf-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, encodingutf-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, encodingdetect_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, encodingutf-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的文件可能要等很久分盘后速度提升非常明显。另外拆完后不要急着删源文件先抽查输出文件的表头、行数和随机几行内容再清理这是我的习惯也建议你保留这个习惯。本文还有配套的精品资源点击获取
返回列表