1. 数据清洗里最不起眼却最要命的一环做数据开发的人都有一个共识真正花时间的地方从来不是写业务逻辑而是把原始数据收拾干净。尤其是当上游数据来自多个业务系统、日志埋点、第三方接口的时候缺失值几乎无处不在。你写一个聚合统计结果发现某个关键字段有30%是null算出来的均值直接偏了你做一个用户画像结果发现年龄字段一半是空的分群逻辑直接失效。SparkSQL作为分布式计算框架里最常用的数据处理工具之一它自带的缺失值处理能力其实比很多人想象的要丰富。但问题在于大部分教程只告诉你dropna和fillna两个方法实际工作中遇到的场景远比这复杂——什么时候该删、什么时候该填、填什么值、怎么填才不破坏数据分布这些决策背后都需要有明确的判断依据。这篇内容面向的是已经用过SparkSQL做基本ETL、但对缺失值处理还停留在“调个API”阶段的开发者。我会从缺失值的类型识别开始讲到不同处理策略的选型逻辑再到SparkSQL中具体函数的实操细节最后分享几个我在实际项目中踩过的坑和总结出来的经验。整篇内容基于Scala和PySpark双语言示例你可以直接对照自己的环境复现。2. 先搞清楚缺失值到底是什么类型2.1 缺失值的三种面孔null、NaN和空字符串很多人一上来就开始写fillna但根本没搞清楚自己数据里的“缺失”到底长什么样。在SparkSQL里缺失值至少有三种不同的表现形式处理方式完全不同。第一种是null这是SparkSQL中最标准的缺失值表示。当DataFrame的某一行某个字段没有值时默认就是null。null在SparkSQL中有特殊的语义——任何与null的运算结果都是null包括null null返回的也是null而不是true。这一点和很多传统数据库的行为一致但在做条件判断时特别容易踩坑。第二种是NaNNot a Number这通常出现在浮点数类型的列中。比如你从CSV读取数据时某个数值列有一个空单元格Spark可能会把它解析为NaN而不是null。NaN和null的区别在于NaN是一个合法的Double类型值它参与运算时不会像null那样“传染”但会导致结果变成NaN。比如1.0 NaN NaN而1.0 null null。第三种是空字符串这在使用concat、split等字符串函数时特别常见。空字符串在SparkSQL中不是null null返回false。很多从Hive表或者JSON日志里读出来的数据缺失值是以空字符串的形式存在的。// Scala示例识别不同类型的缺失值 val df spark.read.option(header, true).csv(path/to/data) // 统计null值 df.filter(col(age).isNull).count() // 统计NaN值 df.filter(col(salary).isNaN).count() // 统计空字符串 df.filter(col(name) ).count()# PySpark示例识别不同类型的缺失值 from pyspark.sql.functions import col, isnan, when, count df spark.read.option(header, true).csv(path/to/data) # 一次性统计所有列的缺失情况 df.select([ count(when(col(c).isNull() | isnan(c) | (col(c) ), c)).alias(c) for c in df.columns ]).show()注意isNaN只能用于数值类型的列对字符串列调用会报错。在实际使用中建议先通过df.dtypes确认列类型再决定用哪种检测方式。2.2 为什么必须先做缺失值画像我见过太多人直接上来就dropna()或者fillna(0)结果要么丢掉了大量有效数据要么引入了严重的偏差。正确的做法是先做一轮“缺失值画像”搞清楚三件事第一每列的缺失比例是多少。如果某列缺失超过80%那这列本身可能就没有保留价值直接drop掉整列比费劲填补更合理。如果缺失在5%以下那删除缺失行的影响很小可以放心drop。第二缺失是随机的还是有规律的。如果缺失值集中在某个时间段、某个业务线、某个用户群体那说明缺失本身携带了信息。比如一个订单表的“优惠券ID”字段大量缺失很可能是因为大部分订单本来就没用优惠券这时候填一个“无优惠券”的标记值比填null更有业务含义。第三缺失值和其他列有没有关联。比如“收入”字段缺失的人可能“职业”字段也大量缺失这说明这些数据可能来自同一个不完整的采集渠道。这种关联性分析可以帮助你判断缺失机制从而选择更合适的处理策略。# 缺失值画像一次性输出每列的缺失统计 from pyspark.sql.functions import col, isnan, when, count, round total df.count() missing_stats df.select([ round(count(when(col(c).isNull() | isnan(c), c)) / total * 100, 2).alias(c _missing_pct) for c in df.columns ]) missing_stats.show()这个画像做完之后你手里就有了一张“缺失地图”接下来每一列该怎么处理心里就有数了。2.3 缺失机制的三种类型及其对处理策略的影响统计学上把缺失机制分为三类虽然听起来偏学术但理解它们对实际决策非常有帮助。完全随机缺失MCAR缺失与任何变量都无关纯粹是随机发生的。比如传感器偶尔丢包导致的数据缺失。这种情况下删除缺失行不会引入偏差直接drop是安全的。随机缺失MAR缺失与其他可观测变量有关但与缺失值本身无关。比如“收入”字段的缺失与“教育程度”有关高学历的人更不愿意透露收入但给定教育程度后收入缺失是随机的。这种情况下可以用其他列的信息来填补比如按教育程度分组后用组内中位数填充。非随机缺失MNAR缺失与缺失值本身有关。比如高收入人群更倾向于不填写收入字段。这种情况下任何填补都会引入偏差最好的做法是保留一个“是否缺失”的标记列让模型自己去学习这个模式。在实际的SparkSQL数据处理中你不需要严格证明数据属于哪种机制但需要有这个意识不是所有缺失值都应该被填补也不是所有缺失行都应该被删除。3. SparkSQL缺失值处理的核心武器库3.1 dropna和fillna的参数细节与常见误用SparkSQL的DataFrameNaFunctions提供了两个核心方法drop和fill在PySpark中对应dropna和fillna。这两个方法的参数设计其实很灵活但很多人只用了最基础的用法。先看drop方法的完整签名// Scala def drop(how: String, cols: Seq[String]): DataFrame def drop(minNonNulls: Int, cols: Seq[String]): DataFrame def drop(how: String): DataFrame def drop(minNonNulls: Int): DataFrame def drop(cols: Seq[String]): DataFrame def drop(): DataFrame# PySpark def dropna(self, howany, threshNone, subsetNone)这里有几个关键参数需要说清楚how可选any或all。any表示只要有一列是null就删除该行all表示所有指定列都是null才删除。默认是any。thresh指定一行中至少要有多少个非null值才保留。这个参数比how更灵活比如你有10列设置thresh8表示至少有8列有值才保留该行。subset指定只考虑哪些列的缺失情况其他列的null不影响删除决策。# 示例只考虑关键列的缺失至少要有3个非空值才保留 df.dropna(thresh3, subset[name, age, city, salary, dept])再看fill方法# PySpark def fillna(self, value, subsetNone) # value可以是单个值、字典列名到填充值的映射// Scala def fill(value: String, cols: Seq[String]): DataFrame def fill(value: Double, cols: Seq[String]): DataFrame def fill(value: Long, cols: Seq[String]): DataFrame def fill(value: Boolean, cols: Seq[String]): DataFrame def fill(valueMap: Map[String, Any]): DataFrame这里有一个非常容易踩的坑fill方法只会填充与给定值类型匹配的列。比如你用fill(0)它只会填充数值类型的列字符串列中的null不会被填充。如果你用fill(unknown)它只会填充字符串列。这个行为在Scala API中尤其明显因为Scala是强类型的。// 这个fill(0)只会影响数值列字符串列的null不受影响 df.na.fill(0) // 如果要同时填充不同类型的列需要用Map df.na.fill(Map( age - 0, name - unknown, is_active - false ))提示在PySpark中fillna的行为略有不同——它会尝试将填充值应用到所有兼容类型的列。但为了精确控制建议始终使用字典形式指定每列的填充值。3.2 用select和when组合实现精细化填充fillna虽然方便但它的填充逻辑是“一刀切”的——要么填固定值要么按列填固定值。实际工作中我们经常需要更精细的控制比如“按分组的中位数填充”“按时间序列的前值填充”“根据其他列的条件动态填充”。这些场景就需要用selectwhenotherwise的组合来实现from pyspark.sql.functions import col, when, median, last from pyspark.sql.window import Window # 场景1按部门分组用组内中位数填充薪资缺失 window_dept Window.partitionBy(dept) median_salary median(salary).over(window_dept) df_filled df.select( name, dept, when(col(salary).isNull(), median_salary) .otherwise(col(salary)) .alias(salary) )// Scala版本 import org.apache.spark.sql.expressions.Window import org.apache.spark.sql.functions._ val windowDept Window.partitionBy(dept) val medianSalary median(salary).over(windowDept) val dfFilled df.select( col(name), col(dept), when(col(salary).isNull, medianSalary) .otherwise(col(salary)) .alias(salary) )这种方式的优势在于填充逻辑完全由你控制可以引用其他列的值、可以调用聚合函数、可以使用窗口函数。缺点是代码量比fillna大但对于生产环境的数据清洗任务来说这点代码量换来的是可解释性和可维护性。3.3 用窗口函数做时序填充last_value和first_value时间序列数据的缺失值处理是一个独立的话题。比如用户行为日志中某个用户某天没有活跃记录但你需要补齐每天的记录用于计算留存率。这时候就不能简单地填0或者填均值而是要用“前一个有效值”来填充。SparkSQL的窗口函数可以很好地解决这个问题from pyspark.sql.functions import last, first, col from pyspark.sql.window import Window # 按用户和时间排序用前一个有效值填充 window_user Window.partitionBy(user_id).orderBy(date) \ .rowsBetween(Window.unboundedPreceding, Window.currentRow) df_filled df.withColumn( value_filled, last(value, ignorenullsTrue).over(window_user) )// Scala版本 val windowUser Window.partitionBy(user_id).orderBy(date) .rowsBetween(Window.unboundedPreceding, Window.currentRow) val dfFilled df.withColumn( value_filled, last(value, ignoreNulls true).over(windowUser) )这里的关键是ignorenulls true参数它让last函数跳过null值取最近的一个非null值。如果不加这个参数遇到null时last会返回null填充就失效了。注意rowsBetween(Window.unboundedPreceding, Window.currentRow)定义了一个从第一行到当前行的窗口。如果你想要“后一个有效值”来填充可以用first配合rowsBetween(Window.currentRow, Window.unboundedFollowing)。3.4 用approxQuantile做分位数填充的完整流程中位数填充是处理数值型缺失值的常用策略因为它对异常值不敏感。SparkSQL提供了approxQuantile方法来近似计算分位数比精确计算中位数快得多尤其适合大数据集。# 计算某列的中位数0.5分位数 median_val df.approxQuantile(salary, [0.5], 0.01)[0] # 用中位数填充 df_filled df.fillna({salary: median_val})// Scala版本 val medianVal df.stat.approxQuantile(salary, Array(0.5), 0.01)(0) val dfFilled df.na.fill(Map(salary - medianVal))approxQuantile的第三个参数是相对误差relative error值越小结果越精确但计算越慢。0.01的误差在大多数场景下已经足够如果数据量特别大可以放宽到0.05。但这里有一个实际项目中经常被忽略的问题全局中位数填充会压缩数据的方差。如果你用全局中位数填充了30%的缺失值那么这30%的数据全部集中在同一个点上后续做统计分析时方差会被严重低估。更好的做法是分组计算中位数比如按地区、按年龄段分组每组用自己的中位数填充。# 分组中位数填充的完整实现 from pyspark.sql.functions import col, median, when, broadcast from pyspark.sql.window import Window # 方法一用窗口函数适合分组数不多的场景 window_spec Window.partitionBy(region) df_filled df.withColumn( salary, when(col(salary).isNull(), median(salary).over(window_spec)) .otherwise(col(salary)) ) # 方法二先计算分组中位数再join回去适合分组数很多的场景 median_df df.groupBy(region).agg(median(salary).alias(median_salary)) df_filled df.join(broadcast(median_df), onregion, howleft) \ .withColumn(salary, when(col(salary).isNull(), col(median_salary)) .otherwise(col(salary)) ).drop(median_salary)方法二在分组数很多比如按城市分组有几百个城市时性能更好因为broadcastjoin把小表分发到每个节点避免了窗口函数的shuffle开销。4. 不同业务场景下的缺失值处理策略选型4.1 用户行为日志缺失往往意味着“没有发生”用户行为日志类数据的缺失值处理有一个特殊之处很多字段的缺失不是数据质量问题而是业务逻辑上的“没有发生”。比如“点击时间”字段为null可能意味着用户根本没有点击“优惠券ID”为null意味着用户没有使用优惠券。这种情况下填0或者填“无”比删除更有业务意义# 用户行为日志的典型处理 df_clean df.fillna({ click_count: 0, # 没有点击就是0次 coupon_id: none, # 没有使用优惠券 stay_duration: 0, # 没有停留就是0秒 is_new_user: False # 默认为老用户 })但这里有一个需要警惕的地方不要把“没有发生”和“数据丢失”混为一谈。如果“点击时间”为null是因为埋点SDK在某些机型上失效导致的那填0就是在掩盖数据质量问题。区分的方法是看缺失的分布——如果缺失集中在某个时间段或某个设备型号那很可能是数据采集问题如果缺失是随机分布的那更可能是业务上的“没有发生”。4.2 交易订单数据金额字段的缺失不能随便填交易类数据中金额字段的缺失是最敏感的。一笔订单的金额为null可能是退款、可能是测试订单、也可能是数据同步延迟。这种情况下删除缺失行往往比填补更安全因为填错金额会直接影响财务报表和业务决策。# 交易数据的保守处理策略 # 1. 先标记缺失保留原始信息 df_marked df.withColumn(amount_missing, col(amount).isNull()) # 2. 对于金额缺失的订单单独输出到异常表供人工核查 df_abnormal df_marked.filter(col(amount_missing) True) df_abnormal.write.saveAsTable(abnormal_orders) # 3. 正常数据继续处理 df_normal df_marked.filter(col(amount_missing) False)如果业务上确认某些金额缺失可以用0填充比如“优惠金额”字段没有优惠就是0那也要在代码中明确注释并且和业务方确认过。4.3 用户画像数据用模型预测填充比统计量填充更合理用户画像场景中缺失值处理的目标和日志、交易数据不同——我们不是要“修正”数据而是要“补全”特征以便后续做推荐或分群。这种情况下用统计量均值、中位数填充虽然简单但会损失特征之间的相关性。更合理的做法是用机器学习模型来预测缺失值。比如“年龄”字段缺失可以用“消费金额”“浏览品类”“活跃时段”等特征训练一个回归模型来预测。SparkML提供了完整的Pipeline支持from pyspark.ml.feature import VectorAssembler from pyspark.ml.regression import RandomForestRegressor from pyspark.ml import Pipeline # 分离有年龄和无年龄的数据 df_with_age df.filter(col(age).isNotNull()) df_without_age df.filter(col(age).isNull()) # 用有年龄的数据训练模型 assembler VectorAssembler( inputCols[purchase_amount, browse_count, active_hours], outputColfeatures ) rf RandomForestRegressor(featuresColfeatures, labelColage) pipeline Pipeline(stages[assembler, rf]) model pipeline.fit(df_with_age) # 预测缺失的年龄 predictions model.transform(df_without_age) df_filled df_with_age.union( predictions.select(user_id, purchase_amount, browse_count, active_hours, col(prediction).alias(age)) )这种方式的代价是计算资源消耗大但如果你要做的是高价值的用户分群或推荐系统这点投入是值得的。4.4 策略选型对照表场景类型推荐策略理由注意事项缺失比例5%直接删除影响小简单可靠确认缺失是MCAR缺失比例5%-30%分组填充保留数据量减少偏差分组字段要有业务意义缺失比例30%考虑删除整列或标记缺失填补可能引入过大偏差评估该列对下游任务的重要性数值型特征中位数/分组中位数对异常值鲁棒注意方差压缩问题类别型特征众数/新增“未知”类别保留类别信息“未知”类别本身可能有预测力时序数据前值填充/插值保持时间连续性注意不要跨用户填充金额/关键指标删除或人工核查填错代价高保留缺失标记供追溯5. 那些文档里不会写的踩坑记录5.1 fillna的类型陷阱为什么我的字符串列没被填充这是我在实际项目中最常遇到的问题之一。Scala API中df.na.fill(0)只会填充数值类型的列字符串列的null完全不受影响。更隐蔽的是如果你的DataFrame中某列是StringType但存储的是数字字符串比如从CSV读进来没做类型转换fill(0)也不会生效。// 假设df有name(String)和age(Integer)两列 // 这个操作只会填充age列的nullname列的null原封不动 df.na.fill(0) // 正确的做法用Map明确指定每列的填充值 df.na.fill(Map(name - unknown, age - 0))在PySpark中fillna(0)的行为是尝试填充所有数值列但如果你传入的是字符串它只会填充字符串列。这个差异在Scala和PySpark之间切换时特别容易搞混。实操建议无论用哪个语言始终用字典/Map形式指定填充值不要依赖默认的类型匹配行为。5.2 dropna的subset参数你以为删的是全部其实只看了部分列dropna的subset参数如果不指定默认会考虑所有列。这意味着只要任意一列有null整行就会被删除。在一个有50列的表里即使每列只有1%的缺失率一行完全没有任何缺失的概率也只有(0.99)^50 ≈ 60%也就是说40%的行会被删掉。# 危险操作不指定subset可能删掉大量数据 df.dropna() # 等价于 df.dropna(howany, subsetNone) # 安全操作只考虑关键列 df.dropna(subset[user_id, order_id, amount])我的经验是永远不要在不指定subset的情况下调用dropna。先想清楚哪些列是“关键列”——这些列的缺失意味着这行数据不可用其他列的缺失可以通过填充解决。5.3 用0填充的隐藏风险均值和方差的双重扭曲用0填充数值型缺失值是最简单粗暴的做法但它的副作用比大多数人想象的要大。假设一个字段的真实均值是100标准差是20你有20%的数据缺失并用0填充那么填充后的均值会变成80标准差会膨胀到约45。这意味着任何基于均值的统计都会偏低任何基于方差的分析都会偏高如果后续做标准化z-score填充的0会被当成一个极端的低值# 对比用0填充 vs 用中位数填充对方差的影响 from pyspark.sql.functions import mean, stddev # 原始数据无缺失 original_stats df.filter(col(value).isNotNull()) \ .agg(mean(value), stddev(value)).collect() # 用0填充 filled_zero df.fillna({value: 0}) zero_stats filled_zero.agg(mean(value), stddev(value)).collect() # 用中位数填充 median_val df.approxQuantile(value, [0.5], 0.01)[0] filled_median df.fillna({value: median_val}) median_stats filled_median.agg(mean(value), stddev(value)).collect()实测下来用中位数填充对方差的影响远小于用0填充。如果业务上确实需要用0填充比如“购买次数”字段那至少要在后续分析中加一个“是否填充”的标记列让模型或分析师知道哪些值是填充出来的。5.4 分布式环境下的approxQuantile精度问题approxQuantile是一个近似算法它的结果依赖于数据的分区方式。在分布式环境中如果数据倾斜严重某些分区的数据量远大于其他分区近似结果的误差可能会超过你设定的relative error。# 检查分区是否均匀 df.rdd.getNumPartitions() df.groupBy(spark_partition_id()).count().show() # 如果数据倾斜先repartition再计算分位数 df_repartitioned df.repartition(200, region) median_val df_repartitioned.approxQuantile(salary, [0.5], 0.01)[0]另外approxQuantile对null值的处理也需要注意——它会自动忽略null值所以计算出的中位数是基于非null数据的。这通常是符合预期的但如果你需要把null也考虑进去比如把null当成0那就要先填充再计算。5.5 多列联合填充时的顺序依赖问题当你需要对多列进行填充且填充逻辑之间存在依赖关系时操作的顺序会影响最终结果。比如你先用“年龄”的中位数填充了年龄缺失然后又想用“年龄”和“收入”的关系来填充收入缺失——这时候填充后的年龄已经不再是原始数据了基于它计算出的收入填充值也会有偏差。# 有依赖关系的填充顺序很重要 # 第一步先填充基础字段 df_step1 df.fillna({age: median_age}) # 第二步基于填充后的age来填充income df_step2 df_step1.withColumn( income, when(col(income).isNull(), col(age) * avg_income_per_age) # 基于age的估算 .otherwise(col(income)) )我的建议是尽量让每列的填充逻辑独立如果确实有依赖关系在代码中明确注释依赖顺序并且对每一步的结果做数据质量检查。6. 把缺失值处理嵌入完整的ETL流水线6.1 设计可复用的缺失值处理函数在实际项目中缺失值处理不应该是散落在各个脚本里的零散代码而应该封装成可复用的函数或类。这样做的价值在于统一处理标准、方便审计和修改、减少重复代码。def clean_missing_values(df, config): 通用的缺失值处理函数 config: { drop_cols: [col1, col2], # 缺失超过阈值直接删除的列 fill_zero: [count1, count2], # 用0填充的列 fill_median: [salary, age], # 用中位数填充的列 fill_mode: [city, category], # 用众数填充的列 fill_custom: {status: unknown}, # 自定义填充 drop_rows_subset: [id, date] # 这些列缺失则删除整行 } from pyspark.sql.functions import col, median, when # 1. 删除整列 if config.get(drop_cols): df df.drop(*config[drop_cols]) # 2. 删除关键列缺失的行 if config.get(drop_rows_subset): df df.dropna(subsetconfig[drop_rows_subset]) # 3. 用0填充 if config.get(fill_zero): fill_dict {c: 0 for c in config[fill_zero]} df df.fillna(fill_dict) # 4. 用中位数填充 if config.get(fill_median): for c in config[fill_median]: median_val df.approxQuantile(c, [0.5], 0.01)[0] df df.fillna({c: median_val}) # 5. 用众数填充 if config.get(fill_mode): for c in config[fill_mode]: mode_val df.groupBy(c).count() \ .orderBy(col(count).desc()).first()[0] df df.fillna({c: mode_val}) # 6. 自定义填充 if config.get(fill_custom): df df.fillna(config[fill_custom]) return df这个函数的设计思路是把“策略”和“执行”分离。配置文件描述每列该怎么处理函数负责执行。这样当业务规则变化时只需要改配置不需要改代码。6.2 处理前后的数据质量校验缺失值处理做完之后必须做一轮校验确认处理结果符合预期。校验的内容包括目标列的缺失率是否降到了预期水平填充后的数据分布是否发生了剧烈变化均值、方差、分位数是否有意外的类型转换比如整数列被填充后变成了浮点列def validate_cleaning(df_before, df_after, target_cols): 对比处理前后的数据质量 from pyspark.sql.functions import mean, stddev, count, isnan, when for c in target_cols: before_missing df_before.filter( col(c).isNull() | isnan(c) ).count() / df_before.count() after_missing df_after.filter( col(c).isNull() | isnan(c) ).count() / df_after.count() before_stats df_before.agg( mean(c).alias(mean), stddev(c).alias(std) ).collect()[0] after_stats df_after.agg( mean(c).alias(mean), stddev(c).alias(std) ).collect()[0] print(f列 {c}:) print(f 缺失率: {before_missing:.2%} - {after_missing:.2%}) print(f 均值: {before_stats[mean]:.2f} - {after_stats[mean]:.2f}) print(f 标准差: {before_stats[std]:.2f} - {after_stats[std]:.2f})如果发现填充后均值偏移超过10%或者标准差变化超过20%就需要重新审视填充策略是否合理。6.3 增量数据处理中的缺失值策略一致性在流式或增量ETL场景中缺失值处理有一个额外的挑战今天用中位数填充明天数据更新后中位数变了历史数据的填充值和新数据的填充值不一致。这会导致同一列的数据在不同批次之间不可比。解决方法是把填充参数持久化。每次计算出的中位数、众数等统计量保存到配置表或文件中后续批次复用同一套参数定期比如每月重新计算并更新。# 保存填充参数 fill_params { salary_median: median_val, age_median: age_median, city_mode: city_mode, computed_date: 2024-01-15 } import json with open(/path/to/fill_params.json, w) as f: json.dump(fill_params, f) # 后续批次读取参数 with open(/path/to/fill_params.json, r) as f: params json.load(f) df df.fillna({salary: params[salary_median]})这个做法看起来简单但在实际项目中经常被忽略导致数据口径不一致的问题很难排查。7. 几个让我印象深刻的真实案例7.1 一个字段的缺失率从5%突然涨到40%某次日常巡检发现用户表的“手机号”字段缺失率从平时的5%突然涨到了40%。第一反应是上游数据同步出了问题但检查了同步任务日志没有发现异常。后来做了一次分组统计发现缺失集中在某个特定渠道的新增用户上。进一步排查发现那个渠道的注册页面最近改版了手机号从必填变成了选填。这不是数据质量问题而是业务规则变化。如果当时直接按“缺失率超过30%就删除整列”的规则处理就会把整个手机号字段删掉影响后续的短信触达和用户召回。这个案例给我的教训是缺失率的变化趋势比缺失率的绝对值更重要。建立缺失率的日常监控比事后补救有效得多。7.2 用均值填充导致的报表偏差一个销售报表中“订单金额”字段有约8%的缺失。当时的处理逻辑是用均值填充。上线一个月后业务方反馈报表中的“总销售额”比财务系统的数字高了约3%。排查后发现缺失的订单金额并不是随机分布的——它们主要来自一个大客户这个客户的订单金额普遍偏高。用全局均值填充后这些订单的金额被低估了但同时又因为填充了值而没有被排除在统计之外导致总销售额出现了偏差。正确的做法应该是对于这种非随机缺失要么删除缺失行并在报表中注明“不含缺失数据”要么用该客户的历史平均订单金额来填充而不是用全局均值。7.3 时序填充时跨用户污染的问题在处理用户日活数据时需要补齐每个用户每天的记录。最初的实现是用last_value按时间排序填充但忘记加partitionBy(user_id)导致用户A的缺失值被用户B的数据填充了。这个bug在测试环境没有发现因为测试数据只有少量用户排序后恰好没有跨用户的情况。修复方法很简单加上partitionBy即可。但这个案例说明时序填充一定要确认窗口函数的partitionBy是否正确否则填充结果完全是错的。# 错误写法没有按用户分区 window_wrong Window.orderBy(date) \ .rowsBetween(Window.unboundedPreceding, Window.currentRow) # 正确写法按用户分区 window_correct Window.partitionBy(user_id).orderBy(date) \ .rowsBetween(Window.unboundedPreceding, Window.currentRow)8. 关于缺失值处理我总结的几条实操原则做了这么多年的数据清洗关于缺失值处理我形成了几个固定的判断原则分享出来供参考。原则一先画像再动手。不要拿到数据就写fillna或dropna。先花10分钟做一轮缺失值统计搞清楚每列的缺失比例、缺失的分布特征、缺失与其他列的关系。这10分钟的投入能帮你避免后面10个小时的返工。原则二能保留标记就保留标记。无论你最终选择删除还是填充都建议在DataFrame中增加一个布尔列标记原始数据是否缺失。这个标记列本身可能就是一个有价值的特征而且在后续排查问题时可以快速定位哪些数据是填充出来的。原则三填充策略要能解释给业务方听。如果你用中位数填充了“收入”字段业务方问起来你要能说清楚为什么用中位数而不是均值、为什么不用0、填充后的数据对分析结论有什么影响。如果解释不清楚说明填充策略可能有问题。原则四处理逻辑要版本化。缺失值处理的配置哪些列删、哪些列填、填什么值应该像代码一样纳入版本管理。每次修改都要有记录方便回溯和审计。原则五不要追求零缺失。有些缺失是数据本身的特性强行填充反而会引入噪声。比如用户画像中的“年收入”字段很多人就是不愿意填你硬要填一个预测值可能还不如保留null让模型自己去处理。最后分享一个我在实际工作中常用的小技巧对于不确定该填什么值的列我会先用不同的策略各跑一遍填0、填中位数、填众数、删除然后对比每种策略下下游核心指标的变化。如果某种策略导致核心指标偏移超过5%就说明这个策略可能不合适。这个方法虽然笨但很有效尤其是在面对不熟悉的业务数据时。