AI数据清洗实践：用Pandas解决脏数据难题

做AI的朋友应该都听过一句话Garbage in garbage out。模型结构再先进训练数据是脏的结果也好不到哪去。在AI基础设施与生态层里看Data数据是整个链条的燃料而数据清洗就是给燃料把关的那个环节。这个事我这些年踩过不少坑也见过太多项目在建模阶段被脏数据坑得死去活来。这篇博文不绕弯子直接把“数据清洗”这件事掰开揉碎讲清楚它为什么关键、到底要处理哪些问题、用什么工具可以做、以及有哪些容易翻车的地方。适合正在搭数据管道的数据工程师、准备做特征工程的算法工程师以及对AI基础设施有兴趣但还没上手接触脏数据的入门同学。1. 为什么说数据清洗是AI基础设施里的关键环节1.1 脏数据比你想的更常见先看几个真实环境里常见的脏数据长什么样APP埋点上报的日志里混着content://com.tencent.wework.fileprovider/external_path/android/data/com...这样的文件路径串网页爬虫抓回来的内容开头是data:text/html;base64,...这种带前缀的完整HTML片段工业传感器回传的数据里一条记录的时间戳是13位毫秒下一条突然变成10位秒还有几条时间戳直接是乱序的数据库导出文件里同一个用户ID在“用户ID”字段下既有数字又有NULL字符串还有一两个空单元格。这些不是段子是我在数据接入阶段反复看到的事。数据处理领域有一句经典的话真实世界的数据永远是脏的。原因不复杂——数据来自不同业务系统、不同采集终端、不同接口版本再加上人为录入错误、网络抖动丢包、日志格式变更数据一汇到一起各种问题全都会浮出来。很多人以为数据清洗只是“写几个正则把乱码过滤掉”实际上远不止这些。在AI基础设施里数据清洗决定了下游特征工程能不能做、模型能不能训得动。数据管道里的第一个环节如果糊弄后面每一步都会放大错误。1.2 不做清洗的代价到底有多大业内流传很广的说法是一个数据科学项目里数据准备和清洗要占掉七八成时间真正建模只占两成。我没精确统计过但从实际项目看这个数字不离谱。如果跳过清洗直接建模最直接的后果是模型性能下降。缺失值会让某些树模型产生偏置异常值会拉偏损失函数重复样本会影响样本权重的分布格式不一致会让特征拼接直接报错。更隐蔽的是训练分布和线上分布不一致的问题——训练时用了带大量异常尖峰的传感器数据模型会把这些尖峰当成正常模式上线后真实数据稍有波动模型就会误判。在AI基础设施层面脏数据还会造成计算资源浪费。我在一个推荐系统项目里见过因为上游日志里出现了大量重复记录每天的清洗任务跑完后特征平台仍然写入了几百万条几乎一样的样本后面训练任务的时间直接翻倍。这种问题如果早期不查排查成本会非常高。1.3 AI基础设施对数据质量提出了新要求过去讲“数据清洗”可能只是数据分析师的日常琐事。但现在AI基础设施已经形成了数据存储、计算引擎、调度系统、特征平台、模型训练和推理服务的完整生态数据清洗不再是可有可无的脚本而是基础设施里的一个标准环节。基础设施要求数据清洗具备三个特性可复用相同的清洗逻辑不能在多个下游里各写一遍要沉淀成公共模块可观测每次清洗执行之后要能记录多少行被过滤、多少字段被填充、质量指标变化如何可回溯一条数据在管道里被改过什么要能通过版本和日志追踪到。这也是为什么现在很多团队会把清洗逻辑放在离在线服务更近的位置做成独立的清洗服务而不是藏在某个模型训练代码里。数据和AI基础设施的关系就像燃料和发动机数据清洗就是燃料精炼工序精炼质量直接决定发动机能不能稳定运转。2. 数据清洗与数据预处理别把概念搞混2.1 清洗和预处理的分工不一样聊数据清洗时总有人把它和数据预处理混着说。我的理解是两者有重叠但目标不同。数据清洗解决的是“数据对不对”的问题重点在修复缺失值补不补、重复值删不删、异常值怎么处理、日期格式统不统一、编码乱不乱、同一实体在不同表里指代是否一致。数据预处理解决的是“数据能不能喂给模型”的问题重点在变换特征标准化、归一化、分箱、类别编码、降维、样本划分这些。通常一条数据处理管道是原始数据进来先做清洗再做预处理。清洗做得干净预处理才有效。如果原始日期字段里混着2024/01/01、2024-01-01、20240101三种格式不先清洗统一后面的时间特征提取就没法做。2.2 数据清洗要管的四类核心问题我把日常数据清洗工作归结成四个大类几乎覆盖了大多数场景。问题类型常见表现典型处理方式缺失值NaN、空字符串、字符串“NULL”、0值占位删除、均值/中位数填充、前向填充、模型预测填充、单独标记重复值整行完全重复、业务主键重复、部分字段重复按业务主键去重保留最新或保留质量最高记录异常值超过业务范围、超过N个标准差、IQR离群点截断、剔除、单独建模、分箱处理格式不一致日期格式混乱、时间戳单位不统一、编码乱码、单位不统一统一格式、统一单位、类型转换、正则规则替换这些只是入口分类。实际业务里四类问题经常叠加在一起。比如一个传感器数据表同一列里既有缺失值又有超过量程的上限值还有重复上报的记录处理顺序就很重要。一般我习惯的顺序是先做格式统一和类型转换再做去重然后处理缺失值最后处理异常值。原因很简单格式没统一时很多“重复值”其实是因为格式不同没被识别出来。2.3 一个具体案例工业传感器数据清洗热词里有“工业传感器数据清洗”这块我专门聊过。传感器数据和其他数据最大的不同是带有时间序列属性清洗时不能仅按单条记录判断还要看上下文。比如温度传感器可能出现一个突变尖峰前一条读数25.4度当前条35.2度下一条又回到25.3度。从单点看35.2并没有超量程上限但从序列看这就是典型的突跳噪声。处理这种问题简单方式是计算当前点与前一点、后一点的差值超过业务阈值就判定为异常然后用滑动窗口的中位数替换。工业场景里还有一个容易踩坑的点传感器数据经常带状态码。同样一个数值状态码是“正常”还是“校准中”可信度完全不同。清洗时不能只盯数值还得把状态码一起考虑不然很容易把正常数据当成异常删掉。3. 用Pandas做数据清洗的完整实操3.1 数据读进来之后先看一眼不要急着动手很多人拿到一个CSV第一件事就是df.dropna()这是最不推荐的。我自己的习惯是清洗之前至少花几分钟做“体检”。import pandas as pd df pd.read_csv(raw_data.csv, encodingutf-8) print(df.shape) print(df.info()) print(df.head(10)) print(df.describe(includeall))df.info()能快速看到每列的非空数量和数据类型df.describe()能看数值列的分布。这两个操作不会修改数据但能帮你判断问题大概集中在哪些列。还有一个容易被忽略的点读文件时最好显式指定dtype和encoding。不指定的话Pandas 可能会把用户ID或手机号识别成数字把有前导零的编码给丢掉。我之前处理过一批订单号因为没指定类型数值型订单号把开头的0全部吞了后面再做关联匹配时全是脏的。3.2 缺失值处理不要无脑 dropna 或 fillna缺失值处理是清洗大头但处理方式要看数据量、缺失比例和业务含义。# 查看每列缺失比例 missing_ratio df.isna().mean() print(missing_ratio) # 缺失比例极低的行可以直接删 df_cleaned df.dropna(subset[user_id, order_time]) # 数值列缺失用中位数填充 df[amount].fillna(df[amount].median(), inplaceTrue) # 时间序列列缺失用前向填充 df[sensor_value].fillna(methodffill, inplaceTrue) # 缺失本身就代表一种状态时单独标记 df[has_amount] df[amount].isna().astype(int)这里有几个经验中位数比均值稳健。数据有偏斜时均值容易被极值带偏直接用均值填充反而会让整体分布失真。时间序列里的缺失优先考虑前向填充或插值而不是用全局均值填充。如果一列缺失比例超过70%除非业务上有明确说法否则我一般会直接丢列。硬填出来的字段不仅没价值还会引入噪声。3.3 去重看清逻辑主键再删去重不是简单地drop_duplicates()关键是先想清楚“重复”是按什么判断的。# 完全重复行 df.drop_duplicates(inplaceTrue) # 按业务主键去重保留最新记录 df.drop_duplicates(subset[user_id, biz_date], keeplast, inplaceTrue) # 保留质量更高的一条比如优先保留备注不为空的行 df[remark_flag] df[remark].notna().astype(int) df.sort_values(remark_flag, ascendingFalse, inplaceTrue) df.drop_duplicates(subset[device_id], keepfirst, inplaceTrue)实际业务里同一个用户一天有多次点击但这不算重复同一张订单在两个系统里各同步了一份才算重复。判断标准完全来自业务口径不能光靠统计学。3.4 异常值处理先圈范围再动刀异常值检测的常用方法有三种业务阈值、Z-score、IQR。# 业务阈值比如商品价格不能为负 df df[df[price] 0] # Z-score超过3个标准差视为异常 from scipy import stats z stats.zscore(df[sensor_value]) df df[(z.abs() 3)] # IQR超出四分位距1.5倍视为异常 Q1 df[amount].quantile(0.25) Q3 df[amount].quantile(0.75) IQR Q3 - Q1 df df[(df[amount] Q1 - 1.5 * IQR) (df[amount] Q3 1.5 * IQR)]Z-score 和 IQR 都只能做粗筛最后还是要结合业务判断。比如看用户年龄时超过3个标准差的记录可能是100岁老人也可能是录入错误直接删除前最好人工抽查几条。我遇到过一种情况某个渠道的客单价天然偏高用全局IQR会把正常的高价值用户当作异常剔除后来改成按渠道分组检测问题才解决。3.5 类型和格式清洗日期、字符串、单位清洗日期和时间戳是高频操作。# 统一日期格式 df[biz_date] pd.to_datetime(df[biz_date], format%Y-%m-%d, errorscoerce) # 处理时间戳13位毫秒 df[ts_ms] pd.to_datetime(df[ts_ms], unitms) # 处理时间戳10位秒 df[ts_s] pd.to_datetime(df[ts_s], units) # 字符串清理去空格、去全角空格、统一大小写 df[user_name] df[user_name].str.strip().str.replace(\u3000, ) df[status] df[status].str.lower() # 单位统一把金额统一为元 df[amount] df[amount].apply(lambda x: x / 100 if x 10000 else x)这里特别提醒pd.to_datetime里加一个errorscoerce解析不了的日期会变成NaT方便下一步统一处理。如果不加解析遇到异常格式时整个程序会直接报错清洗任务就崩了。3.6 文本脏数据清洗正则处理日志和HTML片段现在很多数据清洗任务要处理非结构化文本。比如埋点日志里的完整文件路径、带协议的HTML片段、包含base64的图片串。这种内容不能整条删除而是要按结构抽取有用字段。import re df[raw_text] df[raw_text].astype(str) # 提取形如 content:// 的协议前缀 df[uri_protocol] df[raw_text].str.extract(r(content://[a-zA-Z0-9.])) # 剔除 data:text/html 等前缀保留正文部分 df[html_body] df[raw_text].str.replace(r^data:text/html[^,]*,, , regexTrue) # 去掉URL和文件路径 df[clean_text] df[raw_text].str.replace(r(https?://\S|/storage/emulated/\d/[^\s]), , regexTrue) # 过滤掉纯乱码的base64片段 df[clean_text] df[clean_text].apply(lambda x: re.sub(r[A-Za-z0-9/]{100,}{0,2}, [BASE64], x))这种清洗逻辑没有标准答案完全依赖你对数据源的理解。我的经验是先拿100条样例把能穷举的格式都列出来写正则时一条一条过不要上来就写一个大而全的正则容易被特殊情况打脸。4. 数据清洗规则如何沉淀成自动化工序4.1 清洗规则不能是一次性脚本很多团队最开始就是写一个clean.py跑完就扔。这个思路在短期项目里能应付但一旦数据源更新、业务逻辑变化脚本很快就不可维护了。我更推荐把清洗逻辑拆成函数每个函数只负责一类问题然后统一编排。def clean_duplicates(df): before df.shape[0] df df.drop_duplicates(subset[order_id], keeplast) after df.shape[0] return df, {duplicates_removed: before - after} def clean_missing(df): df[amount] df[amount].fillna(df[amount].median()) return df, {missing_filled: int(df[amount].isna().sum())} def clean_outliers(df): Q1 df[amount].quantile(0.25) Q3 df[amount].quantile(0.75) IQR Q3 - Q1 before df.shape[0] df df[(df[amount] Q1 - 1.5 * IQR) (df[amount] Q3 1.5 * IQR)] return df, {outliers_removed: before - df.shape[0]} def run_data_cleaning(df): reports [] for fn in [clean_duplicates, clean_missing, clean_outliers]: df, report fn(df) reports.append(report) print(reports) return df把每个清洗动作封装成独立函数日后加规则只需新增函数不用动主流程。规则变化时也能通过函数版本记录是谁、在什么时候改的。4.2 数据质量检查清洗前后都要做清洗之后必须检查效果。没有质检的清洗和没洗一样。def quality_report(df): report { total_rows: df.shape[0], total_cols: df.shape[1], missing_ratio: df.isna().mean().to_dict(), duplicate_rows: int(df.duplicated().sum()), numeric_range: {col: [float(df[col].min()), float(df[col].max())] for col in df.select_dtypes(number).columns}, } return report print(清洗前, quality_report(df_raw)) df_cleaned run_data_cleaning(df_raw) print(清洗后, quality_report(df_cleaned))实际生产环境里我会把这份质检报告输出到日志系统或数据库这样调度平台能看到每天清洗的质量变化。如果某天缺失率突然从1%涨到30%大概率是上游数据源出问题了。4.3 工具选型Pandas之外还能用什么数据清洗不只有Pandas。我按场景整理了一份常用工具对比工具适用场景特点Pandas中型数据、快速探索、规则复杂灵活、生态好适合Python团队DataX数据同步过程中做字段过滤和简单转换解耦性好适合离线数据接入管道Pentaho Data Integration图形化ETL、非Python团队无需写代码但大规模处理性能一般OpenRefine探索式清洗、脏数据可视化上手快适合小样本数据调试规则Spark DataFrame海量数据、分布式清洗能横向扩展但代码门槛高我个人的习惯是单机内存放得下的数据优先用Pandas数据量几十GB以上再切Spark。DataX更适合做数据同步不建议拿它做复杂的清洗逻辑它擅长的是把数据从一个地方搬到另一个地方时顺手做点简单转换。4.4 工业级流水线里的位置在实际的AI基础设施流水线里数据清洗的位置通常是在数据接入之后、特征工程之前。离线部分一般是数据源 → DataX同步 → 清洗任务 → 数据质量检查 → 写入数仓/特征库 → 特征工程。在线部分则要求清洗逻辑更精简通常只能做字段校验、格式修正和异常值截断复杂的规则要放到离线的训练数据管道里做。原因是在线推理对延迟敏感不可能为了清洗一整段历史数据而等几秒。还有一个建议清洗规则里的阈值、字段名尽量不要硬编码在代码里可以放到配置中心或规则表里。这样业务改了阈值数据团队不用重新部署代码改个配置就能生效。我在团队里就是把清洗规则统一做成配置文件每天定时拉取管理起来省心很多。5. 常见问题与排查技巧实录5.1 埋点日志里的路径串和HTML片段怎么处理处理APP埋点日志时经常碰到content://...、/storage/emulated/0/android/data/...、data:text/html;base64...这类内容。它们混在业务字段里如果不处理后续做文本特征时会引入大量噪声。我的处理顺序是先把原始值规范化成字符串用正则抽取协议名、路径主目录、文件名后缀等结构化信息如果原始内容对业务无用直接过滤掉如果里面的base64或HTML是业务关注点先解码再清洗解码失败就标记为不可用。一定不要用一条正则就妄图搞定所有情况。埋点格式经常因为SDK版本不同而变化所以给每条清洗规则加上版本号和管理人比追求一次性完美规则更实际。5.2 编码问题和乱码永远排在清洗第一位读文件报UnicodeDecodeError或者是输出后中文变“锟斤拷”这是所有数据清洗新手的老朋友。解决办法# 方案1尝试用可能编码逐个读 for enc in [utf-8, gbk, gb2312, latin1]: try: df pd.read_csv(raw.csv, encodingenc) print(读取成功编码, enc) break except UnicodeDecodeError: continue # 方案2用 chardet 检测编码 import chardet with open(raw.csv, rb) as f: raw f.read(100000) result chardet.detect(raw) df pd.read_csv(raw.csv, encodingresult[encoding])真实项目里从Windows导出的CSV经常是GBK或GB2312而网页爬下来的内容是UTF-8。最好的做法是在清洗脚本里做一次编码检测自动适配来源。5.3 日期时区和时间戳的坑最容易栽跟头我发现很多清洗事故都出在时间戳上。13位毫秒和10位秒混用很常见还有带时区的ISO字符串比如2024-06-01T00:00:0008:00。新手直接用pd.to_datetime转得到的结果常常带时区偏移和另一个不带时区的表做关联时就对不上。我的建议是统一转成UTC时间戳或者统一转成无时区的北京时间。df[event_time] pd.to_datetime(df[event_time], utcTrue) df[event_time_beijing] df[event_time].dt.tz_convert(Asia/Shanghai).dt.tz_localize(None)如果原始数据里既有10位秒又有13位毫秒先判断位数再转换。df[ts_digit_len] df[ts].astype(str).str.len() df[dt] pd.to_datetime(df[ts], units) # 之后再针对13位的重转5.4 空字符串、NaN和None判断起来不是一回事Pandas里None、NaN和空字符串看起来像一回事实际处理完全不同。df.isna()只对NaN和None返回True不算缺失。但真实数据里空字符串往往也代表“没填”。处理办法是把常见“伪缺失”统一转成真正的缺失值。df df.replace(r^\s*$, pd.NA, regexTrue) df df.replace(NULL, pd.NA) df df.replace(null, pd.NA) df df.replace(None, pd.NA)这个步骤最好在读取后立刻执行否则后面的缺失值统计就会失真。5.5 数据清洗里的合规和隐私保护最后说一个容易被忽略的问题清洗过程也会接触到敏感数据。比如手机号、身份证号、地址、健康指标这类字段清洗和调试时经常需要看样例。我的做法是开发环境里先用脱敏后的样例数据不走全量明文清洗脚本中内置一个脱敏函数对手机号、邮箱做掩码处理日志里不要打印完整明细字段只打印行列数和统计指标清洗后的数据落地到仓库时控制访问权限和流转范围。数据清洗不只是技术活它同时也是在给整个AI基础设施建立数据安全的边界。规范做在前面后面能省很多事。我自己在实际操作中还有一个习惯每次清洗任务跑完后除了质量报告顺手把“这次清洗用了哪些规则、每个规则删了多少行、填充了多少值”记到一张统计表里。一个月下来回头翻一翻就能看出上游数据质量在变好还是变坏也能知道该跟哪个团队沟通格式变更。这个过程不需要太复杂一个脚本加一张表就够了。数据清洗这东西做得越细后面建模和上线的日子就越好过。