做数据平台的数据清洗说实话是这个行业里最不受待见、但价值密度最高的活儿。你去看那些搜索热词头歌flume部署、pandas数据处理、MapReduce招聘清洗、网约车Spark清洗、农产品价格清洗……表面上是五花八门的工具和场景实际上全是同一件事把一堆乱七八糟的原始数据变成能支撑分析、算法、报表的干净数据。这篇文章就把这件事掰开揉碎讲清楚适合刚入行的数据开发、数据分析和正在做课程项目/毕业设计的同学也适合所有被脏数据折磨过的人——你会发现很多坑是可以提前避开的。1. 先搞清楚数据平台里的清洗到底在洗什么很多人一听到数据清洗就想到dropna、去重、改个格式这格局就小了。在真实的数据平台里清洗不是孤立写几个函数而是数据从源头进来到最终被消费之间的一道关键工序。你把它放在整个数据链路里看很多决策就自然清晰了。1.1 清洗在数据链路里的位置典型的数据平台有清晰的层次划分ODS操作数据存储原始数据落地层、DWD明细数据层、DWS汇总数据层、ADS应用数据层。清洗主要发生在从ODS到DWD的转换过程中这个位置不是随便定的ODS层尽可能保留原始数据。原因很现实清洗规则可能会变业务方可能会提出新的口径需求如果一进来就洗得面目全非后续想回溯、想重新加工就麻烦了。对于日志数据你甚至要考虑原始日志的完整性。DWD层是清洗后的标准品。这一层要保证字段规范、质量可控、粒度统一下游的汇总、分析、建模都建立在它的基础上。DWD层的质量决定了整个平台的天花板。有人会问那在ETL里清洗和在数仓里清洗有什么区别本质上没区别只是平台给了你更规范的流程位置。清洗逻辑应该沉淀成可复用的脚本或任务挂在调度上而不是每次手动跑一遍临时SQL。1.2 脏数据到底有哪些形态我在实际项目中总结下来脏数据无外乎就这六类每类都能举出活生生的例子缺失值网约车订单表里司机评分字段是空的农产品价格表里部分批发市场没上报当日价格招聘数据里薪资范围缺失一大半。这是最常见、且处理方式最有讲究的一类。重复记录一个用户在业务系统里录入了两次或者因为上游接口重推导致同一张订单在ODS里出现了两遍。校园数据系统里一个学籍信息出现在两张导入表里更是常规操作。异常值招聘数据里薪资写着99999表示面议但没做标记被算法当成真实高薪网约车轨迹数据里出现经度超过180的记录纯属采集器抽风。格式不统一日期有2024-01-01、2024/1/1、20240101三种写法手机号有带86的、带空格的、纯数字的招聘学历有本科、大学本科、本科及以上多个说法。逻辑错误订单的支付时间早于下单时间招聘岗位要求5年经验但学历要求硕士且工作地点显示远程/不限商品价格是负数。无效/过期数据数据平台里经常混入测试数据、软删除数据、以及已完成使命的历史数据。这部分最容易被忽略但它会实实在在污染指标。每类脏数据对应的清洗动作完全不同这就是为什么不能拿着一个dropna走天下。1.3 清洗时机这步同样需要提前想好清洗时机决定了你的整个技术选型。离线 T1 场景下清洗通常放在夜间批处理任务里对延迟不敏感可以用Spark批处理也可以用MapReduce这类离线框架很多教学项目的MapReduce招聘数据清洗就是这个场景。实时场景就完全不一样了如果你是做网约车订单实时风控或者实时大屏清洗逻辑要嵌在流处理管道里用Flink或Spark Streaming处理逻辑要轻量、不能有过重的关联操作。决定时机的主要因素有三个上游数据产生频率、下游消费方对延迟的要求、以及成本。很多团队一上来就想做实时清洗结果发现业务方根本没有实时看数的需求白白增加维护复杂度。我的建议是先满足当前最迫切的业务需求再考虑要不要往实时演进。2. 工具选型pandas、Spark、MapReduce怎么挑搜热词列表里能明显看出一个规律pandas、Spark、MapReduce这三类工具是大家接触最多的。但工具选错后面全靠加班来填坑。根据数据量和场景选工具是一道送分题但很多人在这里丢了分。2.1 一句话选型法数据量在几千到几百万行级别单机内存搞得定需求是探索式清洗、快速迭代pandas绝对首选。数据量上亿、字段多、清洗逻辑复杂需要分布式能力且要跑在集群上Spark用DataFrame API写起来和pandas有相似度但能撑住更大规模。离线批处理、跑在Hadoop生态里且看重任务稳定性和资源可控性MapReduce虽然开发繁琐但可以做到极细粒度的控制教学、考试场景也大量用它来训练数据处理的底层思维。之所以把pandas放第一个推荐是因为太多人忽略了一个事实清洗逻辑的正确性远比清洗逻辑的性能重要。你用Spark洗一亿条数据如果清洗规则本身就是错的那只是在更快地制造垃圾。用pandas先在抽样数据上把规则跑通再迁移到Spark上跑全量这是成本最低、风险最可控的路径。2.2 pandas适用的场景pandas的舒适区就是单机、小规模、高交互。我在做农产品价格数据分析的时候拿到的是各个批发市场发来的Excel和CSV一天也就几千条记录用pandas完全可以而且能一边看结果一边调整规则实时看到哪类脏数据被清理掉、哪些字段还有问题。它的优势在于DataFrame的向量化操作写起来直观代码量是MapReduce的十分之一不到丰富的数据清洗生态isna()、drop_duplicates()、str.strip()、apply()、merge()配合groupby可以做很多依赖上下文的清洗与notebook天然契合清洗过程中的中间结果能被直观地检查和讨论。很多人吐槽pandas处理大数据量会内存爆掉。对这是它的边界。但你不会因为菜市场买菜要备一辆卡车。数据量多大、用什么工具心里得有杆秤。2.3 Spark和MapReduce适用的场景当数据量到了几亿行或者单日增量就超过单机内存就必须上分布式。Spark在这个场景几乎是事实标准——它把复杂的分布式细节藏在了RDD/DataFrame API后面你可以用类似pandas的思维去写清洗逻辑同时获得集群的算力。网约车订单数据清洗用Spark就很合适数据量大、字段多、需要关联地理位置维表和司机维表Spark的分布式join能扛住。MapReduce则是更原始也更可控的方案。实验4招聘数据清洗这类项目选MapReduce核心目的往往是让学习者理解分布式计算的基本原理——map端做什么、reduce端做什么、shuffle是怎么发生的。当你亲手写一个Mapper去解析一行脏数据再写一个Reducer去做去重你会对数据清洗的本质有更深的体感。在实际工程里MapReduce运行慢、开发效率低所以大多数公司已经在用Spark替代它了但理解MapReduce对理解Spark底层原理依然有帮助。2.4 我实际推荐的组合拳我的个人习惯是用pandas做数据探查和规则探索用SQLHive/Spark SQL做常规清洗实在要用代码解决的复杂清洗比如正则解析、自定义UDF再上Spark。绝大多数清洗工作其实用SQL就能完成用SQL的好处是它天然是声明式的同事一看就懂不像一段pandas代码需要逐行解释。把复杂清洗逻辑封装成函数、注册成UDF就兼顾了灵活性和复用性。说到底选型不是比谁的框架高级而是比谁能在合适的数据规模下交出高质量且可维护的结果。3. 一套可以直接抄的清洗流程这部分我拿一个招聘数据的清洗案例来做完整拆解因为招聘数据几乎是所有清洗类型大全有缺失、有重复、有格式乱、有逻辑错误、还有异常值。整个流程分成四步每一步都是我在项目里验证过的。3.1 第一步数据探查与口径确认拿到数据别急着写清洗代码。先回答三个问题数据是什么来源是什么有哪些字段数据量多大哪些字段是核心的、哪些是后面分析或建模要用的业务方对干净数据的定义是什么具体操作上我会先加载一份抽样数据跑这几个pandas命令做快速体检import pandas as pd # 读取原始数据可以指定分隔符、编码 df pd.read_csv(recruitment_raw.csv, encodingutf-8) # 结构性体检数据量、字段名、字段类型、内存占用 df.info() # 数值型字段的分布情况均值、标准差、min、max、缺失个数 df.describe() # 每个字段的缺失值和重复情况 missing_summary df.isna().sum().sort_values(ascendingFalse) print(missing_summary[missing_summary 0]) duplicate_count df.duplicated().sum() print(完全重复行数:, duplicate_count) # 抽样查看若干字段的实际值用眼睛确认“脏”在哪 print(df[[salary_min, salary_max, education, work_year, publish_date]].head(50).to_string())这一步的产出是一份《数据探查报告》明确列出每个字段的质量问题。探查结果可以直接用来和技术负责人、业务方对齐清洗口径。比如薪资字段是保留数值范围还是合并成月薪范围学历字段要不要统一成大专/本科/硕士/博士四个等级这些问题不在动手前确认等你洗完了再返工浪费的时间远超想象。3.2 第二步制定清洗规则基于探查结果把清洗规则写成一张清单。这是整个清洗环节里最关键的一步规则写得越细、越可量化后面执行和验收就越容易。举个招聘数据的例子序号清洗类型字段规则说明1格式统一salary_min / salary_max转为整数型单位统一为千元/月面议记为NULL2缺失处理salary_min / salary_max缺失超过30%且为随机缺失保留字段填充策略用行业均值3去重全字段 job_id按 job_id 去重同一 job_id 保留发布时间最新的一条4异常清洗work_year经验不限转为010年以上转为10超过20的视为异常并置NULL5格式统一education本科及以上、大学本科统一为本科6逻辑校验publish_time发布时间晚于抓取时间或时间为未来时间的记录标记为异常并剔除规则清单要和业务方确认一遍尤其是缺失值填充策略和异常值剔除阈值——这两个最容易引发争议也最需要业务经验来拍板。3.3 第三步编码实现清洗规则规则定了代码就是体力和细心的活。用pandas实现上面的规则大致是这样# 格式统一薪资字段转数值型“面议”置为NaN df[salary_min] pd.to_numeric(df[salary_min], errorscoerce) df[salary_max] pd.to_numeric(df[salary_max], errorscoerce) # 缺失处理薪资缺失用同岗位类型的均值填充 job_group_mean df.groupby(job_category)[salary_max].transform(mean) df[salary_max] df[salary_max].fillna(job_group_mean.round(0)) # 去重按 job_id 去重保留发布时间最新的一行 df df.sort_values(publish_time, ascendingFalse) df df.drop_duplicates(subset[job_id], keepfirst) # 异常清洗工作年限规整 df[work_year] df[work_year].replace(经验不限, 0) df[work_year] df[work_year].str.replace(年以上, , regexFalse) df[work_year] pd.to_numeric(df[work_year], errorscoerce) df[work_year] df[work_year].apply(lambda x: x if (0 x 20) else None) # 逻辑校验剔除发布时间异常的记录 df df[df[publish_time] df[crawl_time]]每一步背后都有明确目的errorscoerce把无法转数值的内容变成NaN不直接报错中断用分组均值填充而不是总体均值是为了尽量贴合同类岗位的真实水平去重前先排序保证保留的是最新一条逻辑校验直接过滤掉未来时间的记录这类记录往往是爬虫抓取时的系统时间错乱导致的。3.4 第四步清洗结果校验与回流清洗代码跑完不代表清洗完成。要做三件事数量校验清洗前X条清洗后Y条Y/X就是清洗通过率。异常偏低的通过率比如低于80%说明上游数据质量问题严重要反馈给采集端。抽样人工检查随机抽50-100条清洗后的数据肉眼检查字段是否符合规则特别是之前出过问题的地方。主键唯一性校验如果是按主键去重的确认去重后主键无重复——这一个校验能拦住大量下游join出重复数据的悲剧。校验通过后把清洗逻辑固化成脚本或调度任务纳入平台的调度系统。这一步现在很多项目里也升级为数据质量稽核自动跑规则、自动报警取代人工抽检。但初期人肉校验依然必要因为机器只能校验你定义过的规则定义之外的问题还得靠人眼。4. 真实项目里那些文档不会写的坑传统的教程会教你函数怎么用但不会告诉你在真实数据平台里同样一个函数用错场景会引发什么后果。这章聊聊我踩过、也看别人踩过的那些坑。4.1 缺失值处理先判断缺失机制再决定怎么补缺失值处理是看起来最简单、实际上最讲究的一步。很多新手拿着fillna(df.mean())一路填下去完全没想过为什么缺。缺失至少分三种完全随机缺失采集设备的偶发故障导致与任何字段无关。这种情况用均值/中位数填充影响较小。随机缺失缺失与否和某些已观测字段相关。招聘数据里薪资缺失往往和岗位类型有关——很多面议岗位集中在高管或特殊工种如果无脑用全局均值填充会严重扭曲工资分布。非随机缺失缺失本身携带信息。比如网约车订单里乘客评分字段缺失可能是因为这个订单根本没被评价也可能是因为订单被取消。这时缺失值本身就是一个业务信号修改评估打分体系前应该先把缺失原因查清楚。所以我的经验是先做缺失模式分析再决定填充策略。用df.isna().mean()看每个字段的缺失比例用相关性分析看缺失是否和其他字段相关。缺失超过40%的字段如果业务上不是核心字段宁可弃用缺失少的字段尽量用更贴近真实的分组填充而不是全局填充。4.2 去重之前先定义什么算重复drop_duplicates()默认是全字段匹配去重但真实业务里你会发现同一笔业务在不同系统里记录的字段并不完全一样。招聘数据里同一个岗位可能在不同招聘平台都有发布岗位内容略有差异但job_id是一样的网约车订单里同一笔订单在一次重试后被重复写入但写入时间字段不一样。所以去重的正确姿势是先明确业务主键再去重。用subset参数指定业务主键字段。同时要考虑时间维度——同一个主键在增量更新场景下保留哪一条我一般会加一个处理时间字段保留最新到达的一条。如果连业务主键都确定不了比如纯日志数据那就退一步定义一个相似度阈值这在极端情况下用文本相似度算法来做但普通业务基本用不到别过度设计。4.3 时间字段和字符编码两个国际化大坑时间字段的坑主要在时区。UC日志、服务器上报的网约车轨迹、跨境电子商务的订单时间字段经常是UTC时间。直接在清洗环节把这个字段当作本地时间处理后续所有时间维度的统计都会出问题。正确的做法是在清洗时就统一到标准时区比如固定转成东八区并确保publish_time这类字段在写入DWD之前已经是带时区信息或已经完成转换的。字符编码则是另一个经典翻车点——尤其那些直接从Windows系统导出的CSV经常是GBK编码pandas默认读UTF-8会报错。我在做农产品价格清洗时遇到过整个文件读进来全是乱码的情况就是因为没注意编码。破局的习惯就是读文件时先明确指定编码encodinggbk或encodingutf-8-sig后者还能处理带BOM头的文件。千万别指望编码自动检测那是个薛定谔的坑。4.4 清洗不是越干净越好这条可能是全篇最重要的经验。很多团队把数据清洗做成了数据暴力清洗——凡是觉得不顺眼的全部剔除最后业务方拿着清洗后的数据一分析发现大量信息丢失指标比实际业务量低了一大截。清洗的核心原则应该是保留有效信息修正错误信息标记异常信息而不是消灭一切看起来不对的记录。对异常数据比较推荐的做法是加一个is_abnormal标记字段保留原始记录的同时标注异常原因而不是直接删除。这样下游既能做全量统计也能筛出异常样本单独分析。数据平台里的清洗永远要给回溯留一条路。另外提醒一句清洗后的数据一定要带上清洗批次号、清洗时间、清洗规则版本这类审计信息。否则出了数据质量问题你连这个数据是用哪版规则洗出来的都查不到复盘就无从谈起。5. 常见问题与排查实录速查表数据清洗的排障本质上就是用最快的方式定位哪个环节不符合预期然后把问题缩小到字段、分区、或某个具体规则上。下面这些是我在实际项目中遇到的高频问题整理成一个速查表直接收藏就行。5.1 高频问题清单现象可能原因排查与解法pandas读CSV直接报错或乱码文件编码不是UTF-8用file命令或notepad查看编码read_csv时指定encoding参数清洗任务内存溢出OOM单机读取数据量过大join操作产生笛卡尔积减小分区/抽样处理检查join字段是否有重复键升级到Spark清洗后数据量骤减异常值过滤条件过严去重主键选择错误打印每一步操作前后的行数定位在哪一步骤减复核规则join或merge之后重复行暴涨join键不是唯一键上游存在重复数据先对上游做唯一性校验确认join语义一对一、一对多日期字段差8小时时区未统一清洗时统一转换为目标时区并在字段注释里写明时区Spark清洗任务数据倾斜某个key的数据量远超其他key考察业务key是否有热点如热门岗位加盐或重新设计key清洗规则对部分数据没生效规则写死在子集上未覆盖所有分支增加规则覆盖率的自动校验用规则清单逐条抽查清洗后的数据和业务口径对不上业务方与开发对干净的定义不一致清洗规则上线前走评审用样例数据逐条确认5.2 排查心法排查数据问题我一般按这个顺序来先看数据量与环比变化缩小到哪一步出了问题再抽样看具体记录确认是规则问题还是数据本身问题最后看日志和调度记录确认是不是跑了旧版本脚本、或者上游任务失败导致输入数据异常。一个非常实用的习惯是给每个清洗步骤输出一个中间结果表并记录行数、主键数量、关键字段缺失率。这一步多花10分钟后面的排查能少花10小时。数据平台里的清洗任务如果没有这些过程指标出问题时只能靠猜效率极低。另外如果是被别人报告的数据不对第一反应不要急着改逻辑。先问清楚哪个指标不对哪个时间范围和什么对比得出的结论很多时候是看数的人拿错了口径不是清洗的问题。保证使用的数据口径一致本身就是清洗工作的一部分。比如岗位平均薪资是全职岗平均还是含实习岗是算数平均数还是中位数这些口径定义清楚并写进文档比什么都管用。5.3 最后分享一个经验回到标题本身——怎么做数据平台的数据清洗。这个问题的答案说白了就四句话先想清楚在管道里清洗的位置和时机再选对工具然后用一套可执行、可校验的流程把规则落地最后把踩过的坑沉淀成排障手册。我在实际操盘中的最大体会是清洗的本质不是写代码而是理解业务。你越懂一个字段背后的业务含义就越能做对该字段怎么洗的决定。一次成功的清洗后期省下的不只是一个staging层而是整个下游分析体系的绝大部分返工成本。这也是为什么我会建议每个做数据的人都认认真真把一次数据清洗从定义到验收完整走一遍——它给你的收获远不止几个技术函数。
