1. 数据清洗在 SparkSQL 里的真实定位1.1 为什么缺失值处理是绕不开的第一道坎做数据开发的人都有一个共识真实生产环境里的数据几乎没有一份是干净的。日志埋点丢字段、业务库导出时 NULL 混着空字符串、上游接口超时导致整列缺失、多表 JOIN 之后右表匹配不上产生大量 NULL——这些都是家常便饭。我做过一个统计在一个中等规模的数据仓库项目里清洗环节大概占整个 ETL 工作量的 60% 到 70%而缺失值处理又占了清洗环节的一半以上。SparkSQL 之所以成为缺失值处理的主力工具核心原因是它把分布式计算能力和类 SQL 的表达方式结合到了一起。你既可以用DataFrame.na系列 API 做精细控制也可以直接写 SQL 用CASE WHEN、COALESCE、IFNULL这些函数处理还能借助 Spark 的 Catalyst 优化器自动做谓词下推和列裁剪。对于动辄几亿行、几十个字段的数据集来说这种声明式表达 分布式执行的组合比手写 MapReduce 或者用单机 Pandas 硬扛要现实得多。这篇文章面向的是已经写过一些 SparkSQL、但对缺失值处理还没有形成系统方法论的开发者。我会从方案选型讲到具体实现从参数计算讲到踩坑记录尽量把我在实际项目里验证过的东西都摊开讲。不管你是刚接触 Spark 的新人还是已经能写窗口函数的老手应该都能从中找到能直接抄作业的部分。1.2 缺失值在 Spark 里的三种面孔很多人一上来就写df.na.drop()结果发现该删的没删掉该留的被删了。问题往往出在没有搞清楚缺失在 Spark 里到底长什么样。实际上SparkSQL 里的缺失值至少有三种形态处理方式完全不同。第一种是真正的null。这是 SQL 语义上的缺失df.na系列 API 默认只认这一种。判断方式是col.isNull或者col.isNotNull。第二种是空字符串。这在使用 Hive 表或者从 CSV 导入数据时特别常见因为很多导出工具会把 NULL 写成空字符串。Spark 的naAPI 默认不把空字符串当缺失值你得手动处理。第三种是业务意义上的无效值比如数字列里的-1、0、999999或者字符串列里的unknown、N/A、null注意这是字符串不是 null。这些值在类型上完全合法但业务上就是缺失。提示在动手清洗之前先用df.summary()或者自己写聚合统计把每一列的 null 数量、空字符串数量、以及可疑的哨兵值数量都摸一遍。这一步花十分钟能省后面两小时的返工。2. 方案选型删除、填充还是标记2.1 三种策略的适用边界缺失值处理归根结底就三条路删掉、填上、或者标记出来交给下游。选哪条路取决于三个因素——缺失比例、字段重要性和业务容忍度。删除策略适合缺失比例很低一般低于 5%且字段不是核心分析维度的情况。比如一张用户行为表里某个可选埋点字段缺失了 2%直接drop掉这些行对整体分析影响微乎其微。但如果缺失比例到了 30%你还去删行那等于把三分之一的数据扔了样本偏差会非常严重。填充策略是最常用的但填充方式的选择很有讲究。用固定值填比如 0、-1、未知最简单但会引入偏差用统计量填均值、中位数、众数相对合理但会压缩方差用前后值填充ffill/bfill适合时间序列用模型预测填充最精确但成本最高。标记策略是很多团队忽略的一招。与其纠结填什么值不如加一个布尔列is_xxx_missing把是否缺失这个信息保留下来原字段该填填、该删删。这样下游做特征工程时缺失本身可能就是一个有预测力的信号。我在一个风控项目里就吃过亏——早期直接把缺失填了均值后来发现某字段缺失和违约率高度相关白白丢了一个强特征。2.2 一张表看清策略选择缺失比例字段角色推荐策略理由 5%非核心直接删除影响小省事 5%核心统计量填充 标记保留样本保留信息5% - 30%非核心固定值填充 标记避免删行造成偏差5% - 30%核心模型填充或分群填充精度要求高 30%任意谨慎评估考虑弃用该字段填充偏差可能大于收益时间序列任意前后值填充保持时间连续性这张表不是铁律但能帮你在没有明确业务指示时快速做决策。实际项目里我一般会先按这个表出一版方案再和业务方对齐。2.3 为什么我不推荐无脑dropna()新手最容易犯的错就是一句df.dropna()走天下。这个操作默认是任意列有 null 就删整行在宽表场景下杀伤力极大。假设一张表有 50 列每列缺失率都是 2%那么一行完全无缺失的概率是 0.98 的 50 次方约等于 0.36。也就是说dropna()之后你只剩三分之一的数据。这个计算很多人没做过一做就明白为什么不能无脑删了。正确的做法是明确指定subset参数只对关键列做删除判断并且用thresh控制容忍度。比如df.na.drop(subset[user_id, event_time], thresh2)表示这两列里至少要有 2 个非空才保留。这样既保证了核心字段的完整性又不会因为边缘字段误伤大量数据。3. 核心 API 与 SQL 函数实操详解3.1DataFrame.na系列 API 的正确用法Spark 的DataFrameNaFunctions提供了drop、fill、replace三个方法覆盖了大部分场景。先说fill它的签名支持传入 Map 做多列不同值填充from pyspark.sql import SparkSession from pyspark.sql.functions import col, when, count spark SparkSession.builder.appName(missing_value_demo).getOrCreate() # 多列分别填充不同值 fill_map { age: 0, city: 未知, income: -1.0 } df_filled df.na.fill(fill_map)这里有个坑fill的 Map 里value 的类型必须和列类型兼容否则 Spark 不会报错而是静默忽略该列。我见过有人给 DoubleType 的列填字符串0结果跑完发现根本没填上排查了半天。所以填充之后一定要用df.filter(col(age).isNull()).count()验证一下。再说replace它常被用来处理哨兵值。比如把-1和999999统一替换成null再走后续填充流程df_cleaned df.na.replace([-1, 999999], [None, None], subset[age, income])注意replace的value参数在新版本里推荐用 dict 形式旧版的 list 形式在列类型不一致时行为诡异。3.2 SQL 风格处理COALESCE与CASE WHEN如果你更习惯写 SQLCOALESCE是最顺手的工具。它返回参数列表中第一个非 null 的值天然适合填充SELECT user_id, COALESCE(city, 未知) AS city, COALESCE(age, 0) AS age, COALESCE(income, avg_income) AS income FROM user_tableCOALESCE的妙处在于它可以级联多个备选值。比如COALESCE(phone, backup_phone, 无联系方式)优先用主手机号没有就用备用号都没有才填默认值。这种降级填充的思路在处理多来源数据合并时特别有用。CASE WHEN则适合更复杂的条件逻辑。比如按用户等级分群填充SELECT user_id, CASE WHEN age IS NOT NULL THEN age WHEN user_level VIP THEN 35 WHEN user_level 普通 THEN 28 ELSE 30 END AS age_filled FROM user_table分群填充比全局均值填充更合理因为不同群体的年龄分布本来就不一样。这个思路在统计上叫条件均值填充能显著降低填充偏差。3.3 空字符串与 null 的统一处理前面说过空字符串是独立于 null 的一种缺失。统一处理的标准做法是先归一化再填充from pyspark.sql.functions import trim, length # 第一步把空字符串和纯空格转成 null for c in string_columns: df df.withColumn( c, when((trim(col(c)) ) | (col(c).isNull()), None).otherwise(col(c)) ) # 第二步统一填充 df df.na.fill({city: 未知, name: 匿名})这里用trim是为了处理 这种纯空格的情况实际数据里这种脏值不少见。归一化之后再填充逻辑就干净了。4. 完整实操流程从原始数据到干净宽表4.1 第一步缺失情况全面体检拿到数据先别急着动手做一次完整体检。我习惯写一个通用的统计函数def missing_report(df, colsNone): cols cols or df.columns total df.count() exprs [] for c in cols: exprs.append( count(when(col(c).isNull() | (trim(col(c).cast(string)) ), c)).alias(c) ) row df.select(exprs).collect()[0] result [] for c in cols: cnt row[c] result.append((c, cnt, round(cnt / total * 100, 2))) return sorted(result, keylambda x: -x[2])这个函数同时统计 null 和空字符串输出按缺失率降序排列。跑一遍之后你就能一眼看出哪些列是重灾区。实测下来一个几千万行的表这个统计大概几十秒能出结果完全可接受。4.2 第二步按列制定处理策略体检报告出来之后逐列过一遍把策略写成一个配置。我强烈建议把策略配置化而不是硬编码在代码里这样后续调整不用改逻辑strategy { user_id: {action: drop, reason: 主键缺失无法关联}, age: {action: fill, value: 0, mark: True}, city: {action: fill, value: 未知}, income: {action: fill, method: median, mark: True}, signup_ts: {action: drop, reason: 时间戳缺失无法分析}, remark: {action: keep, reason: 文本字段缺失无影响} }配置化的好处是当业务方说这个字段的填充值改一下你只需要改配置不用动代码逻辑也方便做版本管理和回溯。4.3 第三步分步执行与验证执行阶段我一般分三步走每步都验证# 步骤1删除关键列缺失的行 drop_cols [c for c, s in strategy.items() if s[action] drop] df df.na.drop(subsetdrop_cols) print(删除后行数:, df.count()) # 步骤2填充 fill_map {c: s[value] for c, s in strategy.items() if s[action] fill and value in s} df df.na.fill(fill_map) # 步骤3中位数等统计量填充 from pyspark.sql.functions import percentile_approx for c, s in strategy.items(): if s.get(method) median: med df.approxQuantile(c, [0.5], 0.01)[0] df df.na.fill({c: med}) # 步骤4加缺失标记列 for c, s in strategy.items(): if s.get(mark): df df.withColumn(fis_{c}_missing, col(c).isNull().cast(int))每一步之后都count()一下确认行数变化符合预期。删除步骤行数应该减少填充步骤行数不变。如果填充后行数变了那一定是哪里写错了。4.4 第四步中位数填充的性能优化approxQuantile是近似算法比精确中位数快很多但要注意它返回的是数组且在大表上仍然可能成为瓶颈。我的经验是如果表超过一亿行先对目标列做采样再算中位数sample_df df.select(income).sample(fraction0.1, seed42) med sample_df.approxQuantile(income, [0.5], 0.01)[0]10% 采样算出来的中位数和全量算出来的差异通常在千分之一以内对填充来说完全够用。这个优化在超大表上能省下大量 shuffle 时间。5. 常见问题与排查技巧实录5.1 填充不生效的几种原因这是被问得最多的问题。填充不生效八成是下面几个原因之一类型不匹配。前面提过fill的 value 类型和列类型不兼容时会静默忽略。解决办法是填充前先df.printSchema()确认类型或者用df.select(c).dtypes单独看。空字符串冒充 null。na.fill不认空字符串你以为的缺失其实是。解决办法是先做归一化把空字符串转成 null。列名拼写错误。Map 的 key 写错了Spark 不报错直接忽略。这个最坑因为完全没提示。解决办法是填充后用isNull().count()逐列验证。缓存导致的假象。如果你对 df 做了 cache然后基于旧 df 做填充看到的可能是缓存数据。解决办法是填充后unpersist再重新读取。5.2 排查速查表现象可能原因排查方法解决fill 后仍有 null类型不匹配printSchema 看类型转换 value 类型fill 后仍有 null空字符串统计 数量先归一化fill 后仍有 null列名错误对比 Map key 和列名修正拼写drop 后行数没变subset 列无 null统计 subset 列 null 数检查数据drop 后行数骤减默认 any 语义检查是否指定 subset加 subset 和 thresh中位数填充报错列含非数值检查列类型先过滤或转换任务 OOM单分区数据倾斜看 Spark UI 分区大小重分区或加盐5.3 几个我踩过的坑坑一dropna的how参数默认是any。这意味着只要有一列是 null 就删行。在宽表上这是灾难。一定要显式指定howall或者用subset。坑二fill对多列用同一个值时如果列类型不同会出问题。比如df.na.fill(0)想给所有数值列填 0但如果有 StringType 的列行为就不确定了。稳妥做法是用 Map 明确指定。坑三时间序列的ffill在 Spark 里没有原生支持。Spark 没有 Pandas 那样的ffill得用窗口函数自己实现from pyspark.sql.window import Window from pyspark.sql.functions import last w Window.orderBy(ts).rowsBetween(Window.unboundedPreceding, Window.currentRow) df df.withColumn(value_filled, last(value, ignorenullsTrue).over(w))这个写法用last配合ignorenullsTrue取当前行之前最后一个非空值效果等同于ffill。注意窗口必须按时间排序且rowsBetween要覆盖到当前行。坑四填充后的数据分布变了下游统计会失真。填充均值会压缩方差填充 0 会拉低均值。所以填充之后最好重新跑一遍描述性统计和填充前对比评估偏差有多大。如果偏差超过可接受范围就得考虑换填充策略。6. 进阶让缺失值处理更聪明的几个思路6.1 基于业务规则的分群填充全局均值填充的最大问题是忽略了群体差异。一个电商用户表如果按用户等级分群VIP 用户的消费均值可能是普通用户的十倍。这时候用全局均值填 VIP 用户的缺失消费额会严重低估。分群填充的实现思路是先按关键维度分组算统计量再 JOIN 回去填充from pyspark.sql.functions import avg # 按用户等级算分组均值 group_stats df.groupBy(user_level).agg(avg(income).alias(avg_income)) # JOIN 回去填充 df_filled df.join(group_stats, onuser_level, howleft) \ .withColumn(income, coalesce(col(income), col(avg_income))) \ .drop(avg_income)这个方案比全局填充精确得多代价是多一次 JOIN。在维度基数不大的情况下比如用户等级就几档性能完全可接受。6.2 用模型预测填充对精度要求极高的场景可以用模型预测缺失值。常见做法是把无缺失的样本作为训练集缺失字段作为标签其他字段作为特征训练一个回归或分类模型再预测缺失样本。Spark MLlib 提供了完整的 Pipeline实现起来不算复杂from pyspark.ml.regression import RandomForestRegressor from pyspark.ml.feature import VectorAssembler # 拆分有标签和无标签样本 train df.filter(col(income).isNotNull()) predict_set df.filter(col(income).isNull()) # 组装特征 assembler VectorAssembler(inputCols[age, tenure, order_cnt], outputColfeatures) train assembler.transform(train) predict_set assembler.transform(predict_set) # 训练预测 rf RandomForestRegressor(featuresColfeatures, labelColincome) model rf.fit(train) pred model.transform(predict_set)模型填充的精度通常比统计量填充高但成本也高而且有过拟合风险。我的建议是只在核心字段、且缺失比例适中10% 到 30%时使用其他情况用分群填充性价比更高。6.3 缺失标记的工程价值前面反复提到加标记列这里展开说说为什么。在机器学习场景里缺失本身往往携带信息。比如一个用户没有填写职业字段可能和他是学生或自由职业相关一个订单没有优惠券 ID可能说明这单没用券。这些信息如果被填充覆盖掉就永久丢失了。加标记列的成本很低就是多一个 int 列但收益可能很大。我在特征工程阶段的标准做法是所有做过填充的字段都加一个对应的is_xxx_missing标记然后让模型自己去判断这个标记有没有用。特征重要性分析会告诉你答案。6.4 处理流程的工程化封装零散的清洗代码很难维护我一般会封装成一个可复用的类class MissingValueHandler: def __init__(self, strategy): self.strategy strategy def fit(self, df): self.stats {} for c, s in self.strategy.items(): if s.get(method) median: self.stats[c] df.approxQuantile(c, [0.5], 0.01)[0] elif s.get(method) mean: self.stats[c] df.select(avg(c)).collect()[0][0] return self def transform(self, df): # 删除 drop_cols [c for c, s in self.strategy.items() if s[action] drop] if drop_cols: df df.na.drop(subsetdrop_cols) # 填充 fill_map {} for c, s in self.strategy.items(): if s[action] fill: fill_map[c] s.get(value, self.stats.get(c)) if fill_map: df df.na.fill(fill_map) # 标记 for c, s in self.strategy.items(): if s.get(mark): df df.withColumn(fis_{c}_missing, col(c).isNull().cast(int)) return df这种 fit/transform 的接口设计和 sklearn 一致好处是统计量在 fit 阶段算一次transform 可以复用到训练集、测试集、线上数据保证填充逻辑一致。这个一致性在模型上线时特别重要否则训练时用训练集均值、预测时用测试集均值会造成数据泄露。7. 性能调优与生产环境注意事项7.1 减少 shuffle 的几个技巧缺失值处理里最耗性能的操作是drop和涉及聚合的填充。drop本身不 shuffle但会改变分区数据量可能造成倾斜。填充里的approxQuantile和groupBy都会触发 shuffle。减少 shuffle 的核心思路是能广播就广播。分群填充时如果分组统计结果很小比如几百行可以用 broadcast joinfrom pyspark.sql.functions import broadcast df_filled df.join(broadcast(group_stats), onuser_level, howleft)广播之后小表分发到每个 executor避免了大表的 shuffle性能提升非常明显。我在一个十亿行的表上做过对比broadcast 之后耗时从 40 分钟降到 8 分钟。7.2 分区与倾斜处理如果某几个分区的数据量远大于其他分区任务会卡在少数几个 task 上。处理缺失值时倾斜往往来自分组维度本身分布不均。比如按城市分组一线城市的数据量可能是小城市的几百倍。解决办法是对倾斜的 key 加随机后缀打散算两阶段聚合。不过对于缺失值填充这种场景更简单的办法是直接改用全局统计量放弃分群精度换取性能。具体怎么权衡看业务对精度的要求。7.3 生产环境的幂等性清洗任务通常要定期跑幂等性很重要。同一个批次的数据跑两次结果应该一致。这就要求填充用的统计量要么固定比如写死在配置里要么从固定的基准表读取不能每次跑都重新算。我见过因为每次重算均值导致下游报表数字对不上的案例排查了很久才发现是填充值在变。提示生产环境的填充统计量建议落一张基准表定期更新而不是每次任务重算。这样既保证幂等又能追溯填充值的来源。7.4 数据质量监控清洗不是一锤子买卖得有监控。我一般会在清洗前后各跑一次缺失率统计把结果写入监控表设置阈值告警。如果某天某个字段的缺失率突然从 2% 飙到 50%那多半是上游出问题了得赶紧排查而不是让清洗任务默默把它填掉。监控指标至少包括各列缺失率、清洗前后行数变化、填充值分布、标记列的比例。这些指标存下来时间长了还能看出数据质量的趋势。8. 我在实际项目中的几点体会做了这么多数据清洗的项目最大的体会是缺失值处理没有银弹只有权衡。删除会丢样本填充会引入偏差标记会增加复杂度每种选择都有代价。关键是想清楚业务能接受什么而不是追求技术上的完美。另一个体会是清洗逻辑一定要可配置、可追溯、可复现。硬编码在脚本里的清洗逻辑过三个月自己都看不懂。配置化之后不仅好维护还能做 A/B 对比——同一份数据用不同策略清洗看下游效果差异用数据说话。最后分享一个小技巧清洗任务上线前一定拿一小份样本数据在本地跑通全流程把每一步的中间结果都打印出来看一遍。我见过太多因为一个类型不匹配或者列名拼写错误导致线上跑了几小时才发现填充没生效的案例。本地十分钟的验证能省线上几小时的返工。数据清洗这件事慢就是快。
