在真实的数据分析任务中,我们拿到的CSV往往不是教科书里那样规整的表格。很多业务系统导出的文件前面有几行制表说明,末尾又附上统计时间和免责声明,中间还可能夹杂空行。如果直接交给pandas读取,轻则数据错位,重则直接抛出异常。掌握一套稳定的清洗与读取流程,是后续建模和报表的前提。

为什么不能直接读取脏CSV
pandas的read_csv默认把第一行当作表头,然后按行解析后续内容。当文件开头存在冗余文本时,这些说明文字会被误认为列名,导致后续所有数据的字段映射全部错误。例如某份销售报表前两句是“本表为2023年区域销售汇总”和“数据来源:内部ERP”,这两行一旦被读成header,真实字段就整体下移,数值列可能变成字符串,统计函数全部失效。
另外,末尾的脚注行通常列数少于正文,pandas在默认模式下会报字段数量不匹配的错误,或者启用错误修正逻辑填NaN,污染数据集。冗余文本还会干扰数据类型推断,本来应该是整数的ID列,因为混入了文字说明而被整体识别为object,后续需要做大量类型转换。因此,先识别并剥离冗余层,再正式建表,是更稳妥的做法。
方案对比:读取前跳过与读入后清洗
第一种思路是在读取阶段就告诉pandas忽略哪些行。read_csv提供了skiprows参数,可以传一个行号列表,也可以传一个函数,对每一行索引返回True表示跳过。对于固定格式的导出文件,比如前3行是说明、后2行是脚注,直接用列表非常高效,不需要把无用数据加载进内存。
第二种思路是把整个文件当成纯文本读进来,用Python做行级过滤,再把干净的内容包装成StringIO交给read_csv。这种方式更灵活,能处理不规则的冗余分布,比如空行随机出现、说明文字位置不固定。缺点是占用更多内存,但在一般规模的数据集上完全可以接受。下面用代码展示两种方式的差异。
import pandas as pd
from io import StringIO
# 方式一:读取前跳过固定行
# 假设前3行是说明,最后2行是脚注,文件共100行
df_skip = pd.read_csv('dirty.csv', skiprows=[0,1,2], skipfooter=2, engine='python')
# 方式二:读入后清洗
with open('dirty.csv', 'r', encoding='utf-8') as f:
lines = f.readlines()
clean_lines = []
for line in lines:
strip_line = line.strip()
# 去掉空行和包含特定说明文字的行
if not strip_line:
continue
if '本表为' in strip_line or '数据来源' in strip_line:
continue
clean_lines.append(line)
clean_text = ''.join(clean_lines)
df_clean = pd.read_csv(StringIO(clean_text))
用正则处理不规则冗余文本
实际工作中,冗余文本常常没有固定行号。比如有的文件在任意位置插入了“注:以上数据为估算值”这样的提示。此时可以用正则表达式在行级别做判断。我们可以编译一个模式,匹配以“注:”开头或全为连字符的分隔行,将其过滤掉。这种写法把清洗规则和业务语义绑定,后续更换文件只需调整正则。
下面的函数演示了如何封装一个通用的清洗读取器。它接收文件路径和过滤正则,返回干净的DataFrame。注意在打开文件时显式指定编码,避免Windows环境下默认gbk导致的中文乱码。清洗完之后,建议打印df.dtypes确认核心列类型符合预期,如果仍有异常,可手动在read_csv中指定dtype参数。
import pandas as pd
import re
from io import StringIO
def read_clean_csv(path, pattern=r'^(注:|--)'):
regex = re.compile(pattern)
with open(path, 'r', encoding='utf-8') as f:
kept = []
for line in f:
if regex.match(line.strip()):
continue
kept.append(line)
return pd.read_csv(StringIO(''.join(kept)))
# 使用示例
df = read_clean_csv('report.csv')
print(df.head())
print(df.dtypes)
利用comment与sep参数增强鲁棒性
如果冗余文本都以某个特殊符号开头,比如每行说明前都有一个井号,那么read_csv的comment参数能直接在解析时忽略该行井号及之后的内容。这比自己写循环更省事,而且是C层面处理,速度更快。不过它只能处理行首或指定起始处的注释,无法应对写在行中间的脏数据。
另外,含冗余文本的文件有时分隔符也不标准,可能是中文逗号或制表符混用。可以在读取时先抽样几行,用csv.Sniffer检测分隔符,再传给sep参数。这样即使文件头部有说明干扰,只要正文规范,依然能正确分列。综合使用comment、skiprows和正则清洗,基本能覆盖绝大多数脏CSV场景。
import pandas as pd
import csv
with open('mixed.csv', 'r', encoding='utf-8') as f:
sample = f.read(2000)
dialect = csv.Sniffer().sniff(sample, delimiters=',t;')
df = pd.read_csv('mixed.csv', comment='#', sep=dialect.delimiter,
skiprows=lambda x: x < 3)
清洗后的校验与落地
无论用哪种方式,清洗完都不能直接信任结果。应当做基础校验:检查行数是否和源文件有效数据吻合,核心字段的空值率是否在合理范围,数值列的最小值最大值有没有明显异常。可以写一个简单的断言函数,在流水线中自动拦截可疑输出。
确认无误后,用to_parquet或to_csv把干净数据持久化,后续分析直接读干净版本,既提升效率也避免重复清洗。如果团队有多人协作,建议把上述读取逻辑做成公共工具模块,统一冗余文本的识别规则,这样才能保证每个人拿到的DataFrame结构一致,减少联调成本。
def validate_df(df, min_rows=10):
assert len(df) >= min_rows, '清洗后数据行数过少'
assert df.isnull().mean().max() < 0.5, '存在空值率过高列'
return True
validate_df(df)
df.to_parquet('clean_report.parquet')