做数据分析的人应该都有这样的经历一份看似完整的报表按用户维度汇总后人数却对不上同一个用户在一张表里出现两次手机号格式还不一样日期字段一部分是2024-01-01另一部分是20240101。真正决定分析质量的往往不是最后那个漂亮的折线图而是进入分析之前的数据清洗。数据清洗做到位数据本身会变得清晰、一致、可解释。这也是“清者自清万人识”在工程里的含义数据是否干净不靠某个人拍脑袋而靠一套可复现的规则让业务、开发和下游模型看到同一份质量结论。这篇文章会围绕一条完整的数据清洗流水线展开。你会看到一份包含重复值、缺失值、异常值、格式不一致和非法手机号的样例数据然后按“读入、探查、修正、校验、落库”五个阶段完成清洗。文章还会给出规则表、清洗报告、常见问题和发布前检查清单方便你直接套用到自己的项目里。1. 数据清洗不是简单删空值而是一套带规则的认知工程数据清洗在很多团队里被理解成“把空值删掉、把重复去掉”。这个理解不够完整。空值和重复只是最外层的问题真正难处理的是同一份数据在不同人眼里有不同含义。比如user_id大小写不一致SQL 里JOIN就会丢数据城市字段中文和英文混写GROUP BY之后会出现两个“北京”。1.1 “清”指的是数据质量可被所有人识别数据清洗的目标不是让数据“看起来顺眼”而是让数据质量可以被量化、被审查、被复现。工程上通常会从这几个维度评估一份数据是否干净质量维度要回答的问题常见脏数据表现完整性关键字段是否缺失手机号为空、订单金额为NULL唯一性是否存在重复记录同一用户出现两次准确性字段值是否真实有效手机号长度不足、金额为负数一致性同一对象在不同表里是否统一user_id大小写混用、城市中英文混写及时性数据是否在预期时间到达日期字段出现未来时间有效性值是否满足业务规则状态字段出现未定义的枚举值“万人识”并不是说所有人都去读原始数据而是指清洗规则、清洗报告、字段定义都是透明的。业务人员可以说“手机号规则是 11 位且以 1 开头”开发人员可以在代码里找到对应规则模型训练时也可以解释为什么某些行被排除。大家看到的是同一套结论而不是各自凭感觉处理数据。1.2 什么时候需要清洗什么时候不要过度清洗清洗不是越多越好。过度清洗会把有效信息当成噪声删掉或者用错误规则填出“看起来很合理”的值。常见场景里这几类数据一般需要清洗场景典型脏数据是否清洗外部 Excel 活动报名表姓名带空格、手机号格式混杂、重复提交需要清洗后入库老系统数据迁移新旧编码不一致、字典值不统一需要清洗并保留映射关系埋点日志解析字段缺失、时间格式混杂、值为null字符串需要清洗后进入分析层数据库直接导出的规范表主键唯一、类型固定只需做基础校验一次性临时查询数据量小分析完即丢弃优先探查不要盲目清洗判断标准很简单如果数据要进入报表、模型、对外接口或者跨团队使用就需要清洗。如果只是临时看一眼清洗规则不明确时先做探查比直接删除更安全。清者自清的前提是先把规则定清楚。2. 环境准备和样例数据集构建为了把清洗流程讲清楚这里用 Python 加 pandas 构建一份模拟的“用户报名表”。数据规模不大适合入门实验也适合作为复杂项目的原型。数据量更大时可以换到 Spark 或数据库 SQL 清洗但核心思路一致。2.1 使用 Python pandas 作为清洗底座pandas 适合处理表格型数据空值、重复值、格式转换都内置了方法。在这类任务里我会额外强调读入时的字段类型设置因为很多脏数据问题是从类型推断开始的。pip install pandas numpy openpyxl如果后续需要读写 Parquet 格式再安装 pyarrowpip install pyarrow如果原始材料没有给出明确版本落地前要确认项目锁定的 Python 和 pandas 版本。下面示例不依赖高阶 APIPython 3.8 以上的常见版本都可以运行。2.2 构造一份包含常见脏数据的样例表先创建一份模拟数据写入raw_users.csv。样例中包含大小写不一致的用户 ID、重复手机号、空手机号、非法手机号、混合日期格式、带空格文本和异常金额。import pandas as pd import numpy as np raw pd.DataFrame({ user_id: [U001, u001, U002, U003, U004, U005, U006, U007, U008, U009], user_name: [ 张三 , 张三, 李四, 王五, 赵六, 孙七, 周八, 吴九, 郑十, 钱十一], city: [北京, 北京, 上海, 广州, 深圳, 杭州, 南京, 武汉, 成都, 西安], phone: [ 13800138000, 13800138000, 13900139000, 15800158000, 123456, 18800000000, None, 17600176000, 13512345678, 19900199000 ], register_date: [ 2024-01-01, 2024/01/01, 2024-01-02, 20240103, 2024-01-05, None, 2024-02-01, 2024-02-02, 2024-02-03, 2024-02-04 ], channel: [ App , APP, Web, H5, APP, 小程序, web, APP, H5, APP], amount: [199.0, 199.0, 58.5, 1001.2, 999999.99, 0.0, 88.8, 66.6, 12.0, 5000.0], status: [success, SUCCESS, failed, success, success, pending, success, success, failed, success] }) raw.to_csv(raw_users.csv, indexFalse, encodingutf-8-sig)这份数据刻意融入了几个典型问题。user_id是U001和u001规范后是重复用户。phone列存在空值、重复值和123456这种明显非法值。register_date混用了-、/和无分隔符格式。channel有前后空格和大小写差异。amount中存在999999.99这种超出常规范围的异常值。注意示例中的手机号和用户均为演示数据不能直接当成线上数据使用。实际项目要把这类表名、字段名和隐私字段脱敏规则替换成自己的。3. 清洗主流程按“读入-探查-修正-校验-落库”五步走清洗最忌讳直接对原始文件动手。正确顺序是先固定字段类型读入再系统探查然后统一修正接着做校验标记最后把清洗结果和报告一起落库。这个顺序能避免“改完发现读错了类型全部白做”。3.1 读入数据时先固定字段类型不要全部交给 pandas 猜测手机号如果被读成整数会丢失前导零用户 ID 如果被读成整数U001这类值会被转成错误内容。所以读入时要明确指定dtype。import pandas as pd import re df pd.read_csv( raw_users.csv, dtype{ user_id: str, user_name: str, city: str, phone: str, channel: str, status: str }, encodingutf-8-sig )这里把字符串字段全部显式声明为str避免 pandas 自动推断。phone即使看起来像数字也按字符串处理。读取时还要注意编码用utf-8-sig而不是utf-8因为保存 CSV 时使用的utf-8-sig会写入 BOM 头之后用 Excel 打开也不容易乱码。3.2 先探查再动手完成空值、重复值、异常值盘点清洗之前先掌握数据全貌。下面代码查看字段类型、空值数量、重复 ID 和城市分布。print(df.info()) print(空值数量) print(df.isna().sum()) print(重复 user_id 数量) print(df[user_id].duplicated().sum()) duplicated_users df[df[user_id].duplicated(keepFalse)] print(重复 user_id 的全部记录) print(duplicated_users) print(城市分布) print(df[city].value_counts(dropnaFalse))从样例数据可看到phone有 1 个空值register_date有 1 个空值user_id表面看没有重复但U001和u001会被当成两个值。如果直接按user_id去重就会漏掉这个重复。这也说明去重前必须先统一格式。异常值的探查不能只依赖describe()。数值字段可以用describe()看最大值和分位数但业务上是否异常要结合业务阈值判断。print(df[amount].describe())amount的最大值会显示为999999.99明显偏离大多数订单金额。此时不要直接删除先标记成异常再交给业务确认。3.3 用统一规则修正标识、文本、日期、金额先做格式统一再做去重。顺序很重要先清洗user_id再drop_duplicates才能把大小写导致的重复识别出来。def normalize_id(value): if pd.isna(value): return value return str(value).strip().upper() def normalize_text(value): if pd.isna(value): return value return re.sub(r\s, , str(value)).strip() def normalize_channel(value): if pd.isna(value): return value return str(value).strip().lower() def normalize_status(value): if pd.isna(value): return value return str(value).strip().lower() df[user_id] df[user_id].map(normalize_id) df[user_name] df[user_name].map(normalize_text) df[channel] df[channel].map(normalize_channel) df[status] df[status].map(normalize_status) df[register_date] pd.to_datetime(df[register_date], errorscoerce)register_date列在转换后可能产生NaT这不是错误而是标记异常日期。errorscoerce的作用是让无法解析的值变成空值而不是抛出异常方便后续统一处理。日期统一之后再处理重复df df.drop_duplicates(subset[user_id], keepfirst).copy()为什么保留第一条而不是最后一条因为原始数据的写入顺序通常代表到达顺序第一条往往更接近用户首次提交。如果业务规则是“保留最新状态”则应该改成keeplast。这个选择必须在清洗报告里说明否则下游无法判断。3.4 保留异常标记而不是“假干净”清洗的常见误区是把异常行直接删掉。更好的做法是新增标记列把“数据有异常”这件事保留下来。比如手机号是否合法、金额是否异常都可以用布尔列标识。def is_valid_phone(value): if pd.isna(value): return False phone re.sub(r\D, , str(value)) return bool(re.fullmatch(r1[3-9]\d{9}, phone)) df[phone_valid] df[phone].map(is_valid_phone) amount_lower 0 amount_upper 100000 df[amount_outlier] ~df[amount].between(amount_lower, amount_upper) print(df[ [user_id, phone, phone_valid, amount, amount_outlier, register_date] ].to_string())手机号校验规则是示例实际项目要根据业务区域和运营商号段调整。比如有些海外手机号长度不同就不能用中国手机号规则一刀切。金额阈值也要来自业务不能只靠统计分位数拍脑袋。保留标记列之后下游可以使用“仅合法手机号且非异常金额”的用户做分析同时还能看到被排除的原因。这样做数据分析仍然是“万人识”每个人都知道某一行为什么不被采用。4. 用一张清洗规则表管理规则而不是散落在代码里小项目里清洗逻辑可以直接写在脚本里但多人协作或数据进入生产环境后散落的if-else很难审查。你不知道哪条规则是谁加的参数为什么这么设。清洗规则表可以解决这个问题。4.1 为什么规则表比 if-else 更好维护规则表本质上是把代码里的判断逻辑抽成配置。业务人员能读规则表开发人员能按规则实现测试人员能按规则造数据。规则表必须包含至少四部分规则编号、字段、规则类型、规则参数和动作。规则编号字段规则类型规则参数动作R001user_id标准化strip upper覆盖原字段R002phone格式校验正则^1[3-9]\d{9}$生成 phone_validR003amount范围校验0 ~ 100000生成 amount_outlierR004register_date日期解析errorscoerce覆盖原字段R005channel标准化strip lower覆盖原字段R006user_id去重keep first删除重复行规则表的好处是清洗流程可审计。如果业务说“手机号规则不能限制为中国手机号”只需要改规则表不用翻代码。4.2 规则表设计与最小执行引擎规则表可以存在 JSON、YAML 或数据库配置表中。这里用 JSON 示例{ rules: [ { id: R001, field: user_id, type: normalize, params: { method: strip_upper }, action: overwrite }, { id: R002, field: phone, type: regex_check, params: { pattern: ^1[3-9]\\d{9}$ }, action: flag, flag_column: phone_valid }, { id: R003, field: amount, type: range_check, params: { min: 0, max: 100000 }, action: flag, flag_column: amount_outlier } ] }对应的最小执行引擎可以按规则类型分发def apply_regex_check(df, rule): pattern rule[params][pattern] df[rule[flag_column]] df[rule[field]].astype(str).str.fullmatch(pattern) return df def apply_range_check(df, rule): min_value rule[params][min] max_value rule[params][max] df[rule[flag_column]] df[rule[field]].between(min_value, max_value) return df这里保留了R002的语义规则名为regex_checkflag 列phone_valid表示真值。业务上“合法手机号”为真非法行为假逻辑更直接。注意规则配置化不代表完全去掉代码。规则类型对应的执行函数仍然需要测试尤其是新增规则类型时要同时补充单元测试。5. 用清洗报告验证效果而不是“看上去对了”清洗完成后不能只说“清洗好了”。判断清洗是否成功要靠清洗报告和数据快照。清洗报告要回答三个问题处理前有多少行处理后有多少行每一类问题处理了多少。5.1 清洗前后必须保留可比较的数据快照不要在原文件上原地修改。项目目录里至少保留raw/和clean/两个目录按日期归档。from pathlib import Path Path(archive).mkdir(exist_okTrue) raw.to_csv(archive/raw_users_20250101.csv, indexFalse, encodingutf-8-sig) clean_df df.copy() clean_df.to_csv(archive/clean_users_20250101.csv, indexFalse, encodingutf-8-sig)归档时可以把raw快照放在只读目录或者加上版本号。这样做的好处是下游发现问题后可以回溯源数据判断是清洗规则问题还是源数据发生了变化。5.2 汇总一张清洗报告让业务和开发看到同一份结论清洗报告用一张汇总表表达。示例字段可以这样计算report_rows [ {指标: 原始行数, 数值: len(raw)}, {指标: 清洗后行数, 数值: len(clean_df)}, {指标: 重复 user_id 删除行数, 数值: len(raw) - len(clean_df)}, {指标: 手机号空值数量, 数值: int(raw[phone].isna().sum())}, {指标: 清洗后手机号非法数量, 数值: int((~clean_df[phone_valid]).sum())}, {指标: 金额异常数量, 数值: int(clean_df[amount_outlier].sum())}, {指标: 日期解析失败数量, 数值: int(clean_df[register_date].isna().sum())}, ] report_df pd.DataFrame(report_rows) print(report_df.to_string(indexFalse))这份报告本身也是一份数据需要能被导出和归档。只有清洗报告稳定输出后后续每次调度才能对比规则是否生效、数据质量是变好还是变坏。在生产环境里清洗报告建议同时写入监控系统或数据库表。开发看日志业务看报表运维看告警。这样数据质量变化就能被及时发现而不是等下游报表出错后才排查。6. 常见问题排查实际清洗中很多问题不是规则复杂而是读取、编码、类型和边界条件。以下四类问题出现频率很高。6.1 中文乱码现象读取 CSV 后中文变成乱码或者用 Excel 打开导出的 CSV 时中文显示异常。常见原因读取时指定的编码和文件保存时的编码不一致。CSV 可能由gbk、gb18030或utf-8保存但代码用另一种编码读取。检查方式用编辑器打开 CSV查看文件右下角编码或者在 Python 中分别尝试encodingutf-8、encodingutf-8-sig、encodinggbk。处理建议统一约定写入和读取都使用utf-8-sig。如果源文件是 Excel 另存的 CSV优先尝试gbk。不要在代码里反复切换编码建议在数据接入层做一次统一转换。6.2 手机号明明正常却被正则拦下现象肉眼检查手机号是 11 位数字但正则校验返回False。常见原因字段被读成浮点数13800138000被显示成13800138000.0手机号带有空格、横线或不可见字符Excel 单元格如果以数字保存导出后可能丢失前导零。检查方式打印repr(value)查看实际字符查看字段的dtype是否为object或str。处理建议读入时将手机号指定为str校验前先用re.sub(r\D, , str(value))去掉非数字字符。如果规则要求严格按原始输入校验则需要在清洗报告中保留原始值和规范化值两列。6.3 日期字段全部变成 NaT现象pd.to_datetime()之后整列日期变成NaT。常见原因列中存在完全无法解析的文本或者日期格式被 Excel 转成了序列号比如45292这种表示日期的数值。errorscoerce会把所有无法解析的值转为NaT有时候反而把整列“吞掉”。检查方式查看to_datetime之前的unique()值确认格式打印转换后日期为空的行观察前后差异。处理建议先对日期做探查判断格式是否统一。如果只有少量格式混杂可以用format参数指定主格式如果格式多样可以按样例正则分组后分别解析。不要把errorscoerce当作万能选项它只是避免抛异常不等于日期是对的。6.4 空值一删数据量少了 30%现象按“至少有一个空值就删行”或“删除所有含空值的行”之后有效数据大幅减少。常见原因没有区分关键字段和普通字段。比如phone为空不代表user_id和amount不能用直接整行删除会损失大量信息。检查方式按列统计空值比例而不是只看总行数。必要时查看空值行的其他字段是否完整。处理建议不要全局dropna()。先给字段分优先级核心字段为空才需要处理普通字段为空可以标记或使用默认值。另一个思路是计算行完整度只删除完整度极低的行。min_completeness 0.8 row_null_ratio df.isna().sum(axis1) / df.shape[1] df_filtered df[row_null_ratio (1 - min_completeness)].copy()这个逻辑表示保留至少 80% 字段非空的行。阈值由业务决定写进清洗规则表之后要比“删除所有空值行”更容易解释。7. 最佳实践清洗过程也要可审查、可回滚、可复用清洗流程上线后真正决定成败的不是第一次清洗效果而是后续维护。源数据一变、业务规则一变、下游口径一变清洗代码都要跟着变。可审查、可回滚、可复用是生产环境的基本要求。7.1 原始数据永远保留一份清洗是对原始数据的加工不是对原始数据的销毁。任何时候都不要直接覆盖raw文件也不要让清洗脚本先读原表再写回原表。落库时清洗结果可以写入新表例如dwd_user_clean原始表保持不动。保留原始数据的另一个作用是对账。当清洗后报表数据与旧系统不一致时可以从原始数据重新计算判断是规则变化还是源数据变化。7.2 清洗规则要有唯一编号和生效时间规则编号要进入代码注释、清洗报告和数据库表。比如R001表示用户 ID 标准化R002表示手机号校验。规则上线时要记录生效时间规则下线时要保留历史记录不能直接删除。7.3 清洗代码要写成可测试函数不要把所有清洗逻辑写在脚本最外层。推荐把一个完整流程封装成函数输入原始 DataFrame输出清洗后的 DataFrame 和清洗报告。def clean_user_data(raw_df: pd.DataFrame) - tuple[pd.DataFrame, pd.DataFrame]: df raw_df.copy() df[user_id] df[user_id].map(normalize_id) df[phone_valid] df[phone].map(is_valid_phone) return df, report_df这样写的好处是可以对函数做单元测试给定一行正常数据、一行空值数据、一行非法手机号数据断言输出结果是否符合预期。等到规则越来越多时测试能够帮助你快速定位是哪条规则误伤了数据。7.4 发布前检查清单清洗任务上线前逐项确认以下内容检查项是否完成原始数据已归档且不参与原地覆盖待确认每条清洗规则都有唯一编号和生效时间待确认清洗报告包含清洗前后行数和关键字段质量指标待确认非法数据是标记而不是全部删除待确认日期、金额、手机号字段类型已固定待确认过滤阈值和业务方确认过待确认清洗输出表有独立表名或版本号待确认异常值不会阻断整个清洗任务待确认任务失败时有日志和告警待确认有回滚方式可以重新执行上一版本规则待确认这张清单也适合作为代码评审的检查项。数据清洗不是“一次性写完就结束”而是每一次规则变更都走同样的审查流程。8. 扩展从一次性清洗走向持续数据质量监控清洗流程跑通之后下一步是把“清者自清”变成一种持续状态。不要让数据质量靠人工救火而是让数据质量指标自动上报。8.1 把清洗规则固化成质量校验任务可以把规则表里的规则转成定时任务每天对增量数据执行同样的校验。校验结果写入质量表至少记录三个指标规则编号、校验时间、异常行数。常见实现有两种。一种是自己写定时任务把清洗报告写到数据库表另一种是使用开源数据质量工具把规则定义为 expectation 套件。无论哪种方式核心都是把“手机号是否合法”这类判断变成可观测、可告警的指标。8.2 让下游消费“清洗后版本”而不是各自清洗数据团队经常遇到的问题是报表团队清洗一次算法团队又清洗一次两边口径不一致。更合理的方式是建立统一的数据清洗层把清洗后的结果作为公共数据提供给下游。在数据量不大时pandas 脚本加归档表就够用。数据量大时可以切换到 Spark 做批量清洗实时场景下可以用 Flink 在流上完成标准化和校验。架构可以变但规则表、清洗报告、回滚机制这些思想不会变。数据清洗的终点不是“数据完全正确”而是“数据质量可以被讨论”。当业务、开发、测试都能指着同一份清洗报告说清楚哪里有问题、规则是什么、为什么这样处理时这份数据才真正做到了清者自清万人识。