戰(zhàn)：從類型識別到策略選型與避坑指南)
1. 數(shù)據(jù)清洗里最不起眼卻最要命的一環(huán)做數(shù)據(jù)開發(fā)的人都有一個共識真正花時間的地方從來不是寫業(yè)務(wù)邏輯而是把原始數(shù)據(jù)收拾干凈。尤其是當(dāng)上游數(shù)據(jù)來自多個業(yè)務(wù)系統(tǒng)、日志埋點(diǎn)、第三方接口的時候缺失值幾乎無處不在。你寫一個聚合統(tǒng)計(jì)結(jié)果發(fā)現(xiàn)某個關(guān)鍵字段有30%是null算出來的均值直接偏了你做一個用戶畫像結(jié)果發(fā)現(xiàn)年齡字段一半是空的分群邏輯直接失效。SparkSQL作為分布式計(jì)算框架里最常用的數(shù)據(jù)處理工具之一它自帶的缺失值處理能力其實(shí)比很多人想象的要豐富。但問題在于大部分教程只告訴你dropna和fillna兩個方法實(shí)際工作中遇到的場景遠(yuǎn)比這復(fù)雜——什么時候該刪、什么時候該填、填什么值、怎么填才不破壞數(shù)據(jù)分布這些決策背后都需要有明確的判斷依據(jù)。這篇內(nèi)容面向的是已經(jīng)用過SparkSQL做基本ETL、但對缺失值處理還停留在“調(diào)個API”階段的開發(fā)者。我會從缺失值的類型識別開始講到不同處理策略的選型邏輯再到SparkSQL中具體函數(shù)的實(shí)操細(xì)節(jié)最后分享幾個我在實(shí)際項(xiàng)目中踩過的坑和總結(jié)出來的經(jīng)驗(yàn)。整篇內(nèi)容基于Scala和PySpark雙語言示例你可以直接對照自己的環(huán)境復(fù)現(xiàn)。2. 先搞清楚缺失值到底是什么類型2.1 缺失值的三種面孔null、NaN和空字符串很多人一上來就開始寫fillna但根本沒搞清楚自己數(shù)據(jù)里的“缺失”到底長什么樣。在SparkSQL里缺失值至少有三種不同的表現(xiàn)形式處理方式完全不同。第一種是null這是SparkSQL中最標(biāo)準(zhǔn)的缺失值表示。當(dāng)DataFrame的某一行某個字段沒有值時默認(rèn)就是null。null在SparkSQL中有特殊的語義——任何與null的運(yùn)算結(jié)果都是null包括null null返回的也是null而不是true。這一點(diǎn)和很多傳統(tǒng)數(shù)據(jù)庫的行為一致但在做條件判斷時特別容易踩坑。第二種是NaNNot a Number這通常出現(xiàn)在浮點(diǎn)數(shù)類型的列中。比如你從CSV讀取數(shù)據(jù)時某個數(shù)值列有一個空單元格Spark可能會把它解析為NaN而不是null。NaN和null的區(qū)別在于NaN是一個合法的Double類型值它參與運(yùn)算時不會像null那樣“傳染”但會導(dǎo)致結(jié)果變成NaN。比如1.0 NaN NaN而1.0 null null。第三種是空字符串這在使用concat、split等字符串函數(shù)時特別常見?？兆址赟parkSQL中不是null null返回false。很多從Hive表或者JSON日志里讀出來的數(shù)據(jù)缺失值是以空字符串的形式存在的。// Scala示例識別不同類型的缺失值 val df spark.read.option(header, true).csv(path/to/data) // 統(tǒng)計(jì)null值 df.filter(col(age).isNull).count() // 統(tǒng)計(jì)NaN值 df.filter(col(salary).isNaN).count() // 統(tǒng)計(jì)空字符串 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) # 一次性統(tǒng)計(jì)所有列的缺失情況 df.select([ count(when(col(c).isNull() | isnan(c) | (col(c) ), c)).alias(c) for c in df.columns ]).show()注意isNaN只能用于數(shù)值類型的列對字符串列調(diào)用會報(bào)錯。在實(shí)際使用中建議先通過df.dtypes確認(rèn)列類型再決定用哪種檢測方式。2.2 為什么必須先做缺失值畫像我見過太多人直接上來就dropna()或者fillna(0)結(jié)果要么丟掉了大量有效數(shù)據(jù)要么引入了嚴(yán)重的偏差。正確的做法是先做一輪“缺失值畫像”搞清楚三件事第一每列的缺失比例是多少。如果某列缺失超過80%那這列本身可能就沒有保留價(jià)值直接drop掉整列比費(fèi)勁填補(bǔ)更合理。如果缺失在5%以下那刪除缺失行的影響很小可以放心drop。第二缺失是隨機(jī)的還是有規(guī)律的。如果缺失值集中在某個時間段、某個業(yè)務(wù)線、某個用戶群體那說明缺失本身攜帶了信息。比如一個訂單表的“優(yōu)惠券ID”字段大量缺失很可能是因?yàn)榇蟛糠钟唵伪緛砭蜎]用優(yōu)惠券這時候填一個“無優(yōu)惠券”的標(biāo)記值比填null更有業(yè)務(wù)含義。第三缺失值和其他列有沒有關(guān)聯(lián)。比如“收入”字段缺失的人可能“職業(yè)”字段也大量缺失這說明這些數(shù)據(jù)可能來自同一個不完整的采集渠道。這種關(guān)聯(lián)性分析可以幫助你判斷缺失機(jī)制從而選擇更合適的處理策略。# 缺失值畫像一次性輸出每列的缺失統(tǒng)計(jì) 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()這個畫像做完之后你手里就有了一張“缺失地圖”接下來每一列該怎么處理心里就有數(shù)了。2.3 缺失機(jī)制的三種類型及其對處理策略的影響統(tǒng)計(jì)學(xué)上把缺失機(jī)制分為三類雖然聽起來偏學(xué)術(shù)但理解它們對實(shí)際決策非常有幫助。完全隨機(jī)缺失MCAR缺失與任何變量都無關(guān)純粹是隨機(jī)發(fā)生的。比如傳感器偶爾丟包導(dǎo)致的數(shù)據(jù)缺失。這種情況下刪除缺失行不會引入偏差直接drop是安全的。隨機(jī)缺失MAR缺失與其他可觀測變量有關(guān)但與缺失值本身無關(guān)。比如“收入”字段的缺失與“教育程度”有關(guān)高學(xué)歷的人更不愿意透露收入但給定教育程度后收入缺失是隨機(jī)的。這種情況下可以用其他列的信息來填補(bǔ)比如按教育程度分組后用組內(nèi)中位數(shù)填充。非隨機(jī)缺失MNAR缺失與缺失值本身有關(guān)。比如高收入人群更傾向于不填寫收入字段。這種情況下任何填補(bǔ)都會引入偏差最好的做法是保留一個“是否缺失”的標(biāo)記列讓模型自己去學(xué)習(xí)這個模式。在實(shí)際的SparkSQL數(shù)據(jù)處理中你不需要嚴(yán)格證明數(shù)據(jù)屬于哪種機(jī)制但需要有這個意識不是所有缺失值都應(yīng)該被填補(bǔ)也不是所有缺失行都應(yīng)該被刪除。3. SparkSQL缺失值處理的核心武器庫3.1 dropna和fillna的參數(shù)細(xì)節(jié)與常見誤用SparkSQL的DataFrameNaFunctions提供了兩個核心方法drop和fill在PySpark中對應(yīng)dropna和fillna。這兩個方法的參數(shù)設(shè)計(jì)其實(shí)很靈活但很多人只用了最基礎(chǔ)的用法。先看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)這里有幾個關(guān)鍵參數(shù)需要說清楚how可選any或all。any表示只要有一列是null就刪除該行all表示所有指定列都是null才刪除。默認(rèn)是any。thresh指定一行中至少要有多少個非null值才保留。這個參數(shù)比how更靈活比如你有10列設(shè)置thresh8表示至少有8列有值才保留該行。subset指定只考慮哪些列的缺失情況其他列的null不影響刪除決策。# 示例只考慮關(guān)鍵列的缺失至少要有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)它只會填充數(shù)值類型的列字符串列中的null不會被填充。如果你用fill(unknown)它只會填充字符串列。這個行為在Scala API中尤其明顯因?yàn)镾cala是強(qiáng)類型的。// 這個fill(0)只會影響數(shù)值列字符串列的null不受影響 df.na.fill(0) // 如果要同時填充不同類型的列需要用Map df.na.fill(Map( age - 0, name - unknown, is_active - false ))提示在PySpark中fillna的行為略有不同——它會嘗試將填充值應(yīng)用到所有兼容類型的列。但為了精確控制建議始終使用字典形式指定每列的填充值。3.2 用select和when組合實(shí)現(xiàn)精細(xì)化填充fillna雖然方便但它的填充邏輯是“一刀切”的——要么填固定值要么按列填固定值。實(shí)際工作中我們經(jīng)常需要更精細(xì)的控制比如“按分組的中位數(shù)填充”“按時間序列的前值填充”“根據(jù)其他列的條件動態(tài)填充”。這些場景就需要用selectwhenotherwise的組合來實(shí)現(xiàn)from pyspark.sql.functions import col, when, median, last from pyspark.sql.window import Window # 場景1按部門分組用組內(nèi)中位數(shù)填充薪資缺失 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) )這種方式的優(yōu)勢在于填充邏輯完全由你控制可以引用其他列的值、可以調(diào)用聚合函數(shù)、可以使用窗口函數(shù)。缺點(diǎn)是代碼量比fillna大但對于生產(chǎn)環(huán)境的數(shù)據(jù)清洗任務(wù)來說這點(diǎn)代碼量換來的是可解釋性和可維護(hù)性。3.3 用窗口函數(shù)做時序填充last_value和first_value時間序列數(shù)據(jù)的缺失值處理是一個獨(dú)立的話題。比如用戶行為日志中某個用戶某天沒有活躍記錄但你需要補(bǔ)齊每天的記錄用于計(jì)算留存率。這時候就不能簡單地填0或者填均值而是要用“前一個有效值”來填充。SparkSQL的窗口函數(shù)可以很好地解決這個問題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) )這里的關(guān)鍵是ignorenulls true參數(shù)它讓last函數(shù)跳過null值取最近的一個非null值。如果不加這個參數(shù)遇到null時last會返回null填充就失效了。注意rowsBetween(Window.unboundedPreceding, Window.currentRow)定義了一個從第一行到當(dāng)前行的窗口。如果你想要“后一個有效值”來填充可以用first配合rowsBetween(Window.currentRow, Window.unboundedFollowing)。3.4 用approxQuantile做分位數(shù)填充的完整流程中位數(shù)填充是處理數(shù)值型缺失值的常用策略因?yàn)樗鼘Ξ惓Ｖ挡幻舾?。SparkSQL提供了approxQuantile方法來近似計(jì)算分位數(shù)比精確計(jì)算中位數(shù)快得多尤其適合大數(shù)據(jù)集。# 計(jì)算某列的中位數(shù)0.5分位數(shù) median_val df.approxQuantile(salary, [0.5], 0.01)[0] # 用中位數(shù)填充 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的第三個參數(shù)是相對誤差relative error值越小結(jié)果越精確但計(jì)算越慢。0.01的誤差在大多數(shù)場景下已經(jīng)足夠如果數(shù)據(jù)量特別大可以放寬到0.05。但這里有一個實(shí)際項(xiàng)目中經(jīng)常被忽略的問題全局中位數(shù)填充會壓縮數(shù)據(jù)的方差。如果你用全局中位數(shù)填充了30%的缺失值那么這30%的數(shù)據(jù)全部集中在同一個點(diǎn)上后續(xù)做統(tǒng)計(jì)分析時方差會被嚴(yán)重低估。更好的做法是分組計(jì)算中位數(shù)比如按地區(qū)、按年齡段分組每組用自己的中位數(shù)填充。# 分組中位數(shù)填充的完整實(shí)現(xiàn) from pyspark.sql.functions import col, median, when, broadcast from pyspark.sql.window import Window # 方法一用窗口函數(shù)適合分組數(shù)不多的場景 window_spec Window.partitionBy(region) df_filled df.withColumn( salary, when(col(salary).isNull(), median(salary).over(window_spec)) .otherwise(col(salary)) ) # 方法二先計(jì)算分組中位數(shù)再join回去適合分組數(shù)很多的場景 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)方法二在分組數(shù)很多比如按城市分組有幾百個城市時性能更好因?yàn)閎roadcastjoin把小表分發(fā)到每個節(jié)點(diǎn)避免了窗口函數(shù)的shuffle開銷。4. 不同業(yè)務(wù)場景下的缺失值處理策略選型4.1 用戶行為日志缺失往往意味著“沒有發(fā)生”用戶行為日志類數(shù)據(jù)的缺失值處理有一個特殊之處很多字段的缺失不是數(shù)據(jù)質(zhì)量問題而是業(yè)務(wù)邏輯上的“沒有發(fā)生”。比如“點(diǎn)擊時間”字段為null可能意味著用戶根本沒有點(diǎn)擊“優(yōu)惠券ID”為null意味著用戶沒有使用優(yōu)惠券。這種情況下填0或者填“無”比刪除更有業(yè)務(wù)意義# 用戶行為日志的典型處理 df_clean df.fillna({ click_count: 0, # 沒有點(diǎn)擊就是0次 coupon_id: none, # 沒有使用優(yōu)惠券 stay_duration: 0, # 沒有停留就是0秒 is_new_user: False # 默認(rèn)為老用戶 })但這里有一個需要警惕的地方不要把“沒有發(fā)生”和“數(shù)據(jù)丟失”混為一談。如果“點(diǎn)擊時間”為null是因?yàn)槁顸c(diǎn)SDK在某些機(jī)型上失效導(dǎo)致的那填0就是在掩蓋數(shù)據(jù)質(zhì)量問題。區(qū)分的方法是看缺失的分布——如果缺失集中在某個時間段或某個設(shè)備型號那很可能是數(shù)據(jù)采集問題如果缺失是隨機(jī)分布的那更可能是業(yè)務(wù)上的“沒有發(fā)生”。4.2 交易訂單數(shù)據(jù)金額字段的缺失不能隨便填交易類數(shù)據(jù)中金額字段的缺失是最敏感的。一筆訂單的金額為null可能是退款、可能是測試訂單、也可能是數(shù)據(jù)同步延遲。這種情況下刪除缺失行往往比填補(bǔ)更安全因?yàn)樘铄e金額會直接影響財(cái)務(wù)報(bào)表和業(yè)務(wù)決策。# 交易數(shù)據(jù)的保守處理策略 # 1. 先標(biāo)記缺失保留原始信息 df_marked df.withColumn(amount_missing, col(amount).isNull()) # 2. 對于金額缺失的訂單單獨(dú)輸出到異常表供人工核查 df_abnormal df_marked.filter(col(amount_missing) True) df_abnormal.write.saveAsTable(abnormal_orders) # 3. 正常數(shù)據(jù)繼續(xù)處理 df_normal df_marked.filter(col(amount_missing) False)如果業(yè)務(wù)上確認(rèn)某些金額缺失可以用0填充比如“優(yōu)惠金額”字段沒有優(yōu)惠就是0那也要在代碼中明確注釋并且和業(yè)務(wù)方確認(rèn)過。4.3 用戶畫像數(shù)據(jù)用模型預(yù)測填充比統(tǒng)計(jì)量填充更合理用戶畫像場景中缺失值處理的目標(biāo)和日志、交易數(shù)據(jù)不同——我們不是要“修正”數(shù)據(jù)而是要“補(bǔ)全”特征以便后續(xù)做推薦或分群。這種情況下用統(tǒng)計(jì)量均值、中位數(shù)填充雖然簡單但會損失特征之間的相關(guān)性。更合理的做法是用機(jī)器學(xué)習(xí)模型來預(yù)測缺失值。比如“年齡”字段缺失可以用“消費(fèi)金額”“瀏覽品類”“活躍時段”等特征訓(xùn)練一個回歸模型來預(yù)測。SparkML提供了完整的Pipeline支持from pyspark.ml.feature import VectorAssembler from pyspark.ml.regression import RandomForestRegressor from pyspark.ml import Pipeline # 分離有年齡和無年齡的數(shù)據(jù) df_with_age df.filter(col(age).isNotNull()) df_without_age df.filter(col(age).isNull()) # 用有年齡的數(shù)據(jù)訓(xùn)練模型 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) # 預(yù)測缺失的年齡 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)) )這種方式的代價(jià)是計(jì)算資源消耗大但如果你要做的是高價(jià)值的用戶分群或推薦系統(tǒng)這點(diǎn)投入是值得的。4.4 策略選型對照表場景類型推薦策略理由注意事項(xiàng)缺失比例5%直接刪除影響小簡單可靠確認(rèn)缺失是MCAR缺失比例5%-30%分組填充保留數(shù)據(jù)量減少偏差分組字段要有業(yè)務(wù)意義缺失比例30%考慮刪除整列或標(biāo)記缺失填補(bǔ)可能引入過大偏差評估該列對下游任務(wù)的重要性數(shù)值型特征中位數(shù)/分組中位數(shù)對異常值魯棒注意方差壓縮問題類別型特征眾數(shù)/新增“未知”類別保留類別信息“未知”類別本身可能有預(yù)測力時序數(shù)據(jù)前值填充/插值保持時間連續(xù)性注意不要跨用戶填充金額/關(guān)鍵指標(biāo)刪除或人工核查填錯代價(jià)高保留缺失標(biāo)記供追溯5. 那些文檔里不會寫的踩坑記錄5.1 fillna的類型陷阱為什么我的字符串列沒被填充這是我在實(shí)際項(xiàng)目中最常遇到的問題之一。Scala API中df.na.fill(0)只會填充數(shù)值類型的列字符串列的null完全不受影響。更隱蔽的是如果你的DataFrame中某列是StringType但存儲的是數(shù)字字符串比如從CSV讀進(jìn)來沒做類型轉(zhuǎn)換fill(0)也不會生效。// 假設(shè)df有name(String)和age(Integer)兩列 // 這個操作只會填充age列的nullname列的null原封不動 df.na.fill(0) // 正確的做法用Map明確指定每列的填充值 df.na.fill(Map(name - unknown, age - 0))在PySpark中fillna(0)的行為是嘗試填充所有數(shù)值列但如果你傳入的是字符串它只會填充字符串列。這個差異在Scala和PySpark之間切換時特別容易搞混。實(shí)操建議無論用哪個語言始終用字典/Map形式指定填充值不要依賴默認(rèn)的類型匹配行為。5.2 dropna的subset參數(shù)你以為刪的是全部其實(shí)只看了部分列dropna的subset參數(shù)如果不指定默認(rèn)會考慮所有列。這意味著只要任意一列有null整行就會被刪除。在一個有50列的表里即使每列只有1%的缺失率一行完全沒有任何缺失的概率也只有(0.99)^50 ≈ 60%也就是說40%的行會被刪掉。# 危險(xiǎn)操作不指定subset可能刪掉大量數(shù)據(jù) df.dropna() # 等價(jià)于 df.dropna(howany, subsetNone) # 安全操作只考慮關(guān)鍵列 df.dropna(subset[user_id, order_id, amount])我的經(jīng)驗(yàn)是永遠(yuǎn)不要在不指定subset的情況下調(diào)用dropna。先想清楚哪些列是“關(guān)鍵列”——這些列的缺失意味著這行數(shù)據(jù)不可用其他列的缺失可以通過填充解決。5.3 用0填充的隱藏風(fēng)險(xiǎn)均值和方差的雙重扭曲用0填充數(shù)值型缺失值是最簡單粗暴的做法但它的副作用比大多數(shù)人想象的要大。假設(shè)一個字段的真實(shí)均值是100標(biāo)準(zhǔn)差是20你有20%的數(shù)據(jù)缺失并用0填充那么填充后的均值會變成80標(biāo)準(zhǔn)差會膨脹到約45。這意味著任何基于均值的統(tǒng)計(jì)都會偏低任何基于方差的分析都會偏高如果后續(xù)做標(biāo)準(zhǔn)化z-score填充的0會被當(dāng)成一個極端的低值# 對比用0填充 vs 用中位數(shù)填充對方差的影響 from pyspark.sql.functions import mean, stddev # 原始數(shù)據(jù)無缺失 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() # 用中位數(shù)填充 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()實(shí)測下來用中位數(shù)填充對方差的影響遠(yuǎn)小于用0填充。如果業(yè)務(wù)上確實(shí)需要用0填充比如“購買次數(shù)”字段那至少要在后續(xù)分析中加一個“是否填充”的標(biāo)記列讓模型或分析師知道哪些值是填充出來的。5.4 分布式環(huán)境下的approxQuantile精度問題approxQuantile是一個近似算法它的結(jié)果依賴于數(shù)據(jù)的分區(qū)方式。在分布式環(huán)境中如果數(shù)據(jù)傾斜嚴(yán)重某些分區(qū)的數(shù)據(jù)量遠(yuǎn)大于其他分區(qū)近似結(jié)果的誤差可能會超過你設(shè)定的relative error。# 檢查分區(qū)是否均勻 df.rdd.getNumPartitions() df.groupBy(spark_partition_id()).count().show() # 如果數(shù)據(jù)傾斜先repartition再計(jì)算分位數(shù) df_repartitioned df.repartition(200, region) median_val df_repartitioned.approxQuantile(salary, [0.5], 0.01)[0]另外approxQuantile對null值的處理也需要注意——它會自動忽略null值所以計(jì)算出的中位數(shù)是基于非null數(shù)據(jù)的。這通常是符合預(yù)期的但如果你需要把null也考慮進(jìn)去比如把null當(dāng)成0那就要先填充再計(jì)算。5.5 多列聯(lián)合填充時的順序依賴問題當(dāng)你需要對多列進(jìn)行填充且填充邏輯之間存在依賴關(guān)系時操作的順序會影響最終結(jié)果。比如你先用“年齡”的中位數(shù)填充了年齡缺失然后又想用“年齡”和“收入”的關(guān)系來填充收入缺失——這時候填充后的年齡已經(jīng)不再是原始數(shù)據(jù)了基于它計(jì)算出的收入填充值也會有偏差。# 有依賴關(guān)系的填充順序很重要 # 第一步先填充基礎(chǔ)字段 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)) )我的建議是盡量讓每列的填充邏輯獨(dú)立如果確實(shí)有依賴關(guān)系在代碼中明確注釋依賴順序并且對每一步的結(jié)果做數(shù)據(jù)質(zhì)量檢查。6. 把缺失值處理嵌入完整的ETL流水線6.1 設(shè)計(jì)可復(fù)用的缺失值處理函數(shù)在實(shí)際項(xiàng)目中缺失值處理不應(yīng)該是散落在各個腳本里的零散代碼而應(yīng)該封裝成可復(fù)用的函數(shù)或類。這樣做的價(jià)值在于統(tǒng)一處理標(biāo)準(zhǔn)、方便審計(jì)和修改、減少重復(fù)代碼。def clean_missing_values(df, config): 通用的缺失值處理函數(shù) config: { drop_cols: [col1, col2], # 缺失超過閾值直接刪除的列 fill_zero: [count1, count2], # 用0填充的列 fill_median: [salary, age], # 用中位數(shù)填充的列 fill_mode: [city, category], # 用眾數(shù)填充的列 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. 刪除關(guān)鍵列缺失的行 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. 用中位數(shù)填充 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. 用眾數(shù)填充 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這個函數(shù)的設(shè)計(jì)思路是把“策略”和“執(zhí)行”分離。配置文件描述每列該怎么處理函數(shù)負(fù)責(zé)執(zhí)行。這樣當(dāng)業(yè)務(wù)規(guī)則變化時只需要改配置不需要改代碼。6.2 處理前后的數(shù)據(jù)質(zhì)量校驗(yàn)缺失值處理做完之后必須做一輪校驗(yàn)確認(rèn)處理結(jié)果符合預(yù)期。校驗(yàn)的內(nèi)容包括目標(biāo)列的缺失率是否降到了預(yù)期水平填充后的數(shù)據(jù)分布是否發(fā)生了劇烈變化均值、方差、分位數(shù)是否有意外的類型轉(zhuǎn)換比如整數(shù)列被填充后變成了浮點(diǎn)列def validate_cleaning(df_before, df_after, target_cols): 對比處理前后的數(shù)據(jù)質(zhì)量 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 標(biāo)準(zhǔn)差: {before_stats[std]:.2f} - {after_stats[std]:.2f})如果發(fā)現(xiàn)填充后均值偏移超過10%或者標(biāo)準(zhǔn)差變化超過20%就需要重新審視填充策略是否合理。6.3 增量數(shù)據(jù)處理中的缺失值策略一致性在流式或增量ETL場景中缺失值處理有一個額外的挑戰(zhàn)今天用中位數(shù)填充明天數(shù)據(jù)更新后中位數(shù)變了歷史數(shù)據(jù)的填充值和新數(shù)據(jù)的填充值不一致。這會導(dǎo)致同一列的數(shù)據(jù)在不同批次之間不可比。解決方法是把填充參數(shù)持久化。每次計(jì)算出的中位數(shù)、眾數(shù)等統(tǒng)計(jì)量保存到配置表或文件中后續(xù)批次復(fù)用同一套參數(shù)定期比如每月重新計(jì)算并更新。# 保存填充參數(shù) 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) # 后續(xù)批次讀取參數(shù) with open(/path/to/fill_params.json, r) as f: params json.load(f) df df.fillna({salary: params[salary_median]})這個做法看起來簡單但在實(shí)際項(xiàng)目中經(jīng)常被忽略導(dǎo)致數(shù)據(jù)口徑不一致的問題很難排查。7. 幾個讓我印象深刻的真實(shí)案例7.1 一個字段的缺失率從5%突然漲到40%某次日常巡檢發(fā)現(xiàn)用戶表的“手機(jī)號”字段缺失率從平時的5%突然漲到了40%。第一反應(yīng)是上游數(shù)據(jù)同步出了問題但檢查了同步任務(wù)日志沒有發(fā)現(xiàn)異常。后來做了一次分組統(tǒng)計(jì)發(fā)現(xiàn)缺失集中在某個特定渠道的新增用戶上。進(jìn)一步排查發(fā)現(xiàn)那個渠道的注冊頁面最近改版了手機(jī)號從必填變成了選填。這不是數(shù)據(jù)質(zhì)量問題而是業(yè)務(wù)規(guī)則變化。如果當(dāng)時直接按“缺失率超過30%就刪除整列”的規(guī)則處理就會把整個手機(jī)號字段刪掉影響后續(xù)的短信觸達(dá)和用戶召回。這個案例給我的教訓(xùn)是缺失率的變化趨勢比缺失率的絕對值更重要。建立缺失率的日常監(jiān)控比事后補(bǔ)救有效得多。7.2 用均值填充導(dǎo)致的報(bào)表偏差一個銷售報(bào)表中“訂單金額”字段有約8%的缺失。當(dāng)時的處理邏輯是用均值填充。上線一個月后業(yè)務(wù)方反饋報(bào)表中的“總銷售額”比財(cái)務(wù)系統(tǒng)的數(shù)字高了約3%。排查后發(fā)現(xiàn)缺失的訂單金額并不是隨機(jī)分布的——它們主要來自一個大客戶這個客戶的訂單金額普遍偏高。用全局均值填充后這些訂單的金額被低估了但同時又因?yàn)樘畛淞酥刀鴽]有被排除在統(tǒng)計(jì)之外導(dǎo)致總銷售額出現(xiàn)了偏差。正確的做法應(yīng)該是對于這種非隨機(jī)缺失要么刪除缺失行并在報(bào)表中注明“不含缺失數(shù)據(jù)”要么用該客戶的歷史平均訂單金額來填充而不是用全局均值。7.3 時序填充時跨用戶污染的問題在處理用戶日活數(shù)據(jù)時需要補(bǔ)齊每個用戶每天的記錄。最初的實(shí)現(xiàn)是用last_value按時間排序填充但忘記加partitionBy(user_id)導(dǎo)致用戶A的缺失值被用戶B的數(shù)據(jù)填充了。這個bug在測試環(huán)境沒有發(fā)現(xiàn)因?yàn)闇y試數(shù)據(jù)只有少量用戶排序后恰好沒有跨用戶的情況。修復(fù)方法很簡單加上partitionBy即可。但這個案例說明時序填充一定要確認(rèn)窗口函數(shù)的partitionBy是否正確否則填充結(jié)果完全是錯的。# 錯誤寫法沒有按用戶分區(qū) window_wrong Window.orderBy(date) \ .rowsBetween(Window.unboundedPreceding, Window.currentRow) # 正確寫法按用戶分區(qū) window_correct Window.partitionBy(user_id).orderBy(date) \ .rowsBetween(Window.unboundedPreceding, Window.currentRow)8. 關(guān)于缺失值處理我總結(jié)的幾條實(shí)操原則做了這么多年的數(shù)據(jù)清洗關(guān)于缺失值處理我形成了幾個固定的判斷原則分享出來供參考。原則一先畫像再動手。不要拿到數(shù)據(jù)就寫fillna或dropna。先花10分鐘做一輪缺失值統(tǒng)計(jì)搞清楚每列的缺失比例、缺失的分布特征、缺失與其他列的關(guān)系。這10分鐘的投入能幫你避免后面10個小時的返工。原則二能保留標(biāo)記就保留標(biāo)記。無論你最終選擇刪除還是填充都建議在DataFrame中增加一個布爾列標(biāo)記原始數(shù)據(jù)是否缺失。這個標(biāo)記列本身可能就是一個有價(jià)值的特征而且在后續(xù)排查問題時可以快速定位哪些數(shù)據(jù)是填充出來的。原則三填充策略要能解釋給業(yè)務(wù)方聽。如果你用中位數(shù)填充了“收入”字段業(yè)務(wù)方問起來你要能說清楚為什么用中位數(shù)而不是均值、為什么不用0、填充后的數(shù)據(jù)對分析結(jié)論有什么影響。如果解釋不清楚說明填充策略可能有問題。原則四處理邏輯要版本化。缺失值處理的配置哪些列刪、哪些列填、填什么值應(yīng)該像代碼一樣納入版本管理。每次修改都要有記錄方便回溯和審計(jì)。原則五不要追求零缺失。有些缺失是數(shù)據(jù)本身的特性強(qiáng)行填充反而會引入噪聲。比如用戶畫像中的“年收入”字段很多人就是不愿意填你硬要填一個預(yù)測值可能還不如保留null讓模型自己去處理。最后分享一個我在實(shí)際工作中常用的小技巧對于不確定該填什么值的列我會先用不同的策略各跑一遍填0、填中位數(shù)、填眾數(shù)、刪除然后對比每種策略下下游核心指標(biāo)的變化。如果某種策略導(dǎo)致核心指標(biāo)偏移超過5%就說明這個策略可能不合適。這個方法雖然笨但很有效尤其是在面對不熟悉的業(yè)務(wù)數(shù)據(jù)時。 SparkSQL缺失值處理實(shí)戰(zhàn)：從類型識別到策略選型與避坑指南
1. 數(shù)據(jù)清洗里最不起眼卻最要命的一環(huán)做數(shù)據(jù)開發(fā)的人都有一個共識真正花時間的地方從來不是寫業(yè)務(wù)邏輯而是把原始數(shù)據(jù)收拾干凈。尤其是當(dāng)上游數(shù)據(jù)來自多個業(yè)務(wù)系統(tǒng)、日志埋點(diǎn)、第三方接口的時候缺失值幾乎無處不在。你寫一個聚合統(tǒng)計(jì)結(jié)果發(fā)現(xiàn)某個關(guān)鍵字段有30%是null算出來的均值直接偏了你做一個用戶畫像結(jié)果發(fā)現(xiàn)年齡字段一半是空的分群邏輯直接失效。SparkSQL作為分布式計(jì)算框架里最常用的數(shù)據(jù)處理工具之一它自帶的缺失值處理能力其實(shí)比很多人想象的要豐富。但問題在于大部分教程只告訴你dropna和fillna兩個方法實(shí)際工作中遇到的場景遠(yuǎn)比這復(fù)雜——什么時候該刪、什么時候該填、填什么值、怎么填才不破壞數(shù)據(jù)分布這些決策背后都需要有明確的判斷依據(jù)。這篇內(nèi)容面向的是已經(jīng)用過SparkSQL做基本ETL、但對缺失值處理還停留在“調(diào)個API”階段的開發(fā)者。我會從缺失值的類型識別開始講到不同處理策略的選型邏輯再到SparkSQL中具體函數(shù)的實(shí)操細(xì)節(jié)最后分享幾個我在實(shí)際項(xiàng)目中踩過的坑和總結(jié)出來的經(jīng)驗(yàn)。整篇內(nèi)容基于Scala和PySpark雙語言示例你可以直接對照自己的環(huán)境復(fù)現(xiàn)。2. 先搞清楚缺失值到底是什么類型2.1 缺失值的三種面孔null、NaN和空字符串很多人一上來就開始寫fillna但根本沒搞清楚自己數(shù)據(jù)里的“缺失”到底長什么樣。在SparkSQL里缺失值至少有三種不同的表現(xiàn)形式處理方式完全不同。第一種是null這是SparkSQL中最標(biāo)準(zhǔn)的缺失值表示。當(dāng)DataFrame的某一行某個字段沒有值時默認(rèn)就是null。null在SparkSQL中有特殊的語義——任何與null的運(yùn)算結(jié)果都是null包括null null返回的也是null而不是true。這一點(diǎn)和很多傳統(tǒng)數(shù)據(jù)庫的行為一致但在做條件判斷時特別容易踩坑。第二種是NaNNot a Number這通常出現(xiàn)在浮點(diǎn)數(shù)類型的列中。比如你從CSV讀取數(shù)據(jù)時某個數(shù)值列有一個空單元格Spark可能會把它解析為NaN而不是null。NaN和null的區(qū)別在于NaN是一個合法的Double類型值它參與運(yùn)算時不會像null那樣“傳染”但會導(dǎo)致結(jié)果變成NaN。比如1.0 NaN NaN而1.0 null null。第三種是空字符串這在使用concat、split等字符串函數(shù)時特別常見?？兆址赟parkSQL中不是null null返回false。很多從Hive表或者JSON日志里讀出來的數(shù)據(jù)缺失值是以空字符串的形式存在的。// Scala示例識別不同類型的缺失值 val df spark.read.option(header, true).csv(path/to/data) // 統(tǒng)計(jì)null值 df.filter(col(age).isNull).count() // 統(tǒng)計(jì)NaN值 df.filter(col(salary).isNaN).count() // 統(tǒng)計(jì)空字符串 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) # 一次性統(tǒng)計(jì)所有列的缺失情況 df.select([ count(when(col(c).isNull() | isnan(c) | (col(c) ), c)).alias(c) for c in df.columns ]).show()注意isNaN只能用于數(shù)值類型的列對字符串列調(diào)用會報(bào)錯。在實(shí)際使用中建議先通過df.dtypes確認(rèn)列類型再決定用哪種檢測方式。2.2 為什么必須先做缺失值畫像我見過太多人直接上來就dropna()或者fillna(0)結(jié)果要么丟掉了大量有效數(shù)據(jù)要么引入了嚴(yán)重的偏差。正確的做法是先做一輪“缺失值畫像”搞清楚三件事第一每列的缺失比例是多少。如果某列缺失超過80%那這列本身可能就沒有保留價(jià)值直接drop掉整列比費(fèi)勁填補(bǔ)更合理。如果缺失在5%以下那刪除缺失行的影響很小可以放心drop。第二缺失是隨機(jī)的還是有規(guī)律的。如果缺失值集中在某個時間段、某個業(yè)務(wù)線、某個用戶群體那說明缺失本身攜帶了信息。比如一個訂單表的“優(yōu)惠券ID”字段大量缺失很可能是因?yàn)榇蟛糠钟唵伪緛砭蜎]用優(yōu)惠券這時候填一個“無優(yōu)惠券”的標(biāo)記值比填null更有業(yè)務(wù)含義。第三缺失值和其他列有沒有關(guān)聯(lián)。比如“收入”字段缺失的人可能“職業(yè)”字段也大量缺失這說明這些數(shù)據(jù)可能來自同一個不完整的采集渠道。這種關(guān)聯(lián)性分析可以幫助你判斷缺失機(jī)制從而選擇更合適的處理策略。# 缺失值畫像一次性輸出每列的缺失統(tǒng)計(jì) 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()這個畫像做完之后你手里就有了一張“缺失地圖”接下來每一列該怎么處理心里就有數(shù)了。2.3 缺失機(jī)制的三種類型及其對處理策略的影響統(tǒng)計(jì)學(xué)上把缺失機(jī)制分為三類雖然聽起來偏學(xué)術(shù)但理解它們對實(shí)際決策非常有幫助。完全隨機(jī)缺失MCAR缺失與任何變量都無關(guān)純粹是隨機(jī)發(fā)生的。比如傳感器偶爾丟包導(dǎo)致的數(shù)據(jù)缺失。這種情況下刪除缺失行不會引入偏差直接drop是安全的。隨機(jī)缺失MAR缺失與其他可觀測變量有關(guān)但與缺失值本身無關(guān)。比如“收入”字段的缺失與“教育程度”有關(guān)高學(xué)歷的人更不愿意透露收入但給定教育程度后收入缺失是隨機(jī)的。這種情況下可以用其他列的信息來填補(bǔ)比如按教育程度分組后用組內(nèi)中位數(shù)填充。非隨機(jī)缺失MNAR缺失與缺失值本身有關(guān)。比如高收入人群更傾向于不填寫收入字段。這種情況下任何填補(bǔ)都會引入偏差最好的做法是保留一個“是否缺失”的標(biāo)記列讓模型自己去學(xué)習(xí)這個模式。在實(shí)際的SparkSQL數(shù)據(jù)處理中你不需要嚴(yán)格證明數(shù)據(jù)屬于哪種機(jī)制但需要有這個意識不是所有缺失值都應(yīng)該被填補(bǔ)也不是所有缺失行都應(yīng)該被刪除。3. SparkSQL缺失值處理的核心武器庫3.1 dropna和fillna的參數(shù)細(xì)節(jié)與常見誤用SparkSQL的DataFrameNaFunctions提供了兩個核心方法drop和fill在PySpark中對應(yīng)dropna和fillna。這兩個方法的參數(shù)設(shè)計(jì)其實(shí)很靈活但很多人只用了最基礎(chǔ)的用法。先看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)這里有幾個關(guān)鍵參數(shù)需要說清楚how可選any或all。any表示只要有一列是null就刪除該行all表示所有指定列都是null才刪除。默認(rèn)是any。thresh指定一行中至少要有多少個非null值才保留。這個參數(shù)比how更靈活比如你有10列設(shè)置thresh8表示至少有8列有值才保留該行。subset指定只考慮哪些列的缺失情況其他列的null不影響刪除決策。# 示例只考慮關(guān)鍵列的缺失至少要有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)它只會填充數(shù)值類型的列字符串列中的null不會被填充。如果你用fill(unknown)它只會填充字符串列。這個行為在Scala API中尤其明顯因?yàn)镾cala是強(qiáng)類型的。// 這個fill(0)只會影響數(shù)值列字符串列的null不受影響 df.na.fill(0) // 如果要同時填充不同類型的列需要用Map df.na.fill(Map( age - 0, name - unknown, is_active - false ))提示在PySpark中fillna的行為略有不同——它會嘗試將填充值應(yīng)用到所有兼容類型的列。但為了精確控制建議始終使用字典形式指定每列的填充值。3.2 用select和when組合實(shí)現(xiàn)精細(xì)化填充fillna雖然方便但它的填充邏輯是“一刀切”的——要么填固定值要么按列填固定值。實(shí)際工作中我們經(jīng)常需要更精細(xì)的控制比如“按分組的中位數(shù)填充”“按時間序列的前值填充”“根據(jù)其他列的條件動態(tài)填充”。這些場景就需要用selectwhenotherwise的組合來實(shí)現(xiàn)from pyspark.sql.functions import col, when, median, last from pyspark.sql.window import Window # 場景1按部門分組用組內(nèi)中位數(shù)填充薪資缺失 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) )這種方式的優(yōu)勢在于填充邏輯完全由你控制可以引用其他列的值、可以調(diào)用聚合函數(shù)、可以使用窗口函數(shù)。缺點(diǎn)是代碼量比fillna大但對于生產(chǎn)環(huán)境的數(shù)據(jù)清洗任務(wù)來說這點(diǎn)代碼量換來的是可解釋性和可維護(hù)性。3.3 用窗口函數(shù)做時序填充last_value和first_value時間序列數(shù)據(jù)的缺失值處理是一個獨(dú)立的話題。比如用戶行為日志中某個用戶某天沒有活躍記錄但你需要補(bǔ)齊每天的記錄用于計(jì)算留存率。這時候就不能簡單地填0或者填均值而是要用“前一個有效值”來填充。SparkSQL的窗口函數(shù)可以很好地解決這個問題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) )這里的關(guān)鍵是ignorenulls true參數(shù)它讓last函數(shù)跳過null值取最近的一個非null值。如果不加這個參數(shù)遇到null時last會返回null填充就失效了。注意rowsBetween(Window.unboundedPreceding, Window.currentRow)定義了一個從第一行到當(dāng)前行的窗口。如果你想要“后一個有效值”來填充可以用first配合rowsBetween(Window.currentRow, Window.unboundedFollowing)。3.4 用approxQuantile做分位數(shù)填充的完整流程中位數(shù)填充是處理數(shù)值型缺失值的常用策略因?yàn)樗鼘Ξ惓Ｖ挡幻舾?。SparkSQL提供了approxQuantile方法來近似計(jì)算分位數(shù)比精確計(jì)算中位數(shù)快得多尤其適合大數(shù)據(jù)集。# 計(jì)算某列的中位數(shù)0.5分位數(shù) median_val df.approxQuantile(salary, [0.5], 0.01)[0] # 用中位數(shù)填充 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的第三個參數(shù)是相對誤差relative error值越小結(jié)果越精確但計(jì)算越慢。0.01的誤差在大多數(shù)場景下已經(jīng)足夠如果數(shù)據(jù)量特別大可以放寬到0.05。但這里有一個實(shí)際項(xiàng)目中經(jīng)常被忽略的問題全局中位數(shù)填充會壓縮數(shù)據(jù)的方差。如果你用全局中位數(shù)填充了30%的缺失值那么這30%的數(shù)據(jù)全部集中在同一個點(diǎn)上后續(xù)做統(tǒng)計(jì)分析時方差會被嚴(yán)重低估。更好的做法是分組計(jì)算中位數(shù)比如按地區(qū)、按年齡段分組每組用自己的中位數(shù)填充。# 分組中位數(shù)填充的完整實(shí)現(xiàn) from pyspark.sql.functions import col, median, when, broadcast from pyspark.sql.window import Window # 方法一用窗口函數(shù)適合分組數(shù)不多的場景 window_spec Window.partitionBy(region) df_filled df.withColumn( salary, when(col(salary).isNull(), median(salary).over(window_spec)) .otherwise(col(salary)) ) # 方法二先計(jì)算分組中位數(shù)再join回去適合分組數(shù)很多的場景 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)方法二在分組數(shù)很多比如按城市分組有幾百個城市時性能更好因?yàn)閎roadcastjoin把小表分發(fā)到每個節(jié)點(diǎn)避免了窗口函數(shù)的shuffle開銷。4. 不同業(yè)務(wù)場景下的缺失值處理策略選型4.1 用戶行為日志缺失往往意味著“沒有發(fā)生”用戶行為日志類數(shù)據(jù)的缺失值處理有一個特殊之處很多字段的缺失不是數(shù)據(jù)質(zhì)量問題而是業(yè)務(wù)邏輯上的“沒有發(fā)生”。比如“點(diǎn)擊時間”字段為null可能意味著用戶根本沒有點(diǎn)擊“優(yōu)惠券ID”為null意味著用戶沒有使用優(yōu)惠券。這種情況下填0或者填“無”比刪除更有業(yè)務(wù)意義# 用戶行為日志的典型處理 df_clean df.fillna({ click_count: 0, # 沒有點(diǎn)擊就是0次 coupon_id: none, # 沒有使用優(yōu)惠券 stay_duration: 0, # 沒有停留就是0秒 is_new_user: False # 默認(rèn)為老用戶 })但這里有一個需要警惕的地方不要把“沒有發(fā)生”和“數(shù)據(jù)丟失”混為一談。如果“點(diǎn)擊時間”為null是因?yàn)槁顸c(diǎn)SDK在某些機(jī)型上失效導(dǎo)致的那填0就是在掩蓋數(shù)據(jù)質(zhì)量問題。區(qū)分的方法是看缺失的分布——如果缺失集中在某個時間段或某個設(shè)備型號那很可能是數(shù)據(jù)采集問題如果缺失是隨機(jī)分布的那更可能是業(yè)務(wù)上的“沒有發(fā)生”。4.2 交易訂單數(shù)據(jù)金額字段的缺失不能隨便填交易類數(shù)據(jù)中金額字段的缺失是最敏感的。一筆訂單的金額為null可能是退款、可能是測試訂單、也可能是數(shù)據(jù)同步延遲。這種情況下刪除缺失行往往比填補(bǔ)更安全因?yàn)樘铄e金額會直接影響財(cái)務(wù)報(bào)表和業(yè)務(wù)決策。# 交易數(shù)據(jù)的保守處理策略 # 1. 先標(biāo)記缺失保留原始信息 df_marked df.withColumn(amount_missing, col(amount).isNull()) # 2. 對于金額缺失的訂單單獨(dú)輸出到異常表供人工核查 df_abnormal df_marked.filter(col(amount_missing) True) df_abnormal.write.saveAsTable(abnormal_orders) # 3. 正常數(shù)據(jù)繼續(xù)處理 df_normal df_marked.filter(col(amount_missing) False)如果業(yè)務(wù)上確認(rèn)某些金額缺失可以用0填充比如“優(yōu)惠金額”字段沒有優(yōu)惠就是0那也要在代碼中明確注釋并且和業(yè)務(wù)方確認(rèn)過。4.3 用戶畫像數(shù)據(jù)用模型預(yù)測填充比統(tǒng)計(jì)量填充更合理用戶畫像場景中缺失值處理的目標(biāo)和日志、交易數(shù)據(jù)不同——我們不是要“修正”數(shù)據(jù)而是要“補(bǔ)全”特征以便后續(xù)做推薦或分群。這種情況下用統(tǒng)計(jì)量均值、中位數(shù)填充雖然簡單但會損失特征之間的相關(guān)性。更合理的做法是用機(jī)器學(xué)習(xí)模型來預(yù)測缺失值。比如“年齡”字段缺失可以用“消費(fèi)金額”“瀏覽品類”“活躍時段”等特征訓(xùn)練一個回歸模型來預(yù)測。SparkML提供了完整的Pipeline支持from pyspark.ml.feature import VectorAssembler from pyspark.ml.regression import RandomForestRegressor from pyspark.ml import Pipeline # 分離有年齡和無年齡的數(shù)據(jù) df_with_age df.filter(col(age).isNotNull()) df_without_age df.filter(col(age).isNull()) # 用有年齡的數(shù)據(jù)訓(xùn)練模型 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) # 預(yù)測缺失的年齡 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)) )這種方式的代價(jià)是計(jì)算資源消耗大但如果你要做的是高價(jià)值的用戶分群或推薦系統(tǒng)這點(diǎn)投入是值得的。4.4 策略選型對照表場景類型推薦策略理由注意事項(xiàng)缺失比例5%直接刪除影響小簡單可靠確認(rèn)缺失是MCAR缺失比例5%-30%分組填充保留數(shù)據(jù)量減少偏差分組字段要有業(yè)務(wù)意義缺失比例30%考慮刪除整列或標(biāo)記缺失填補(bǔ)可能引入過大偏差評估該列對下游任務(wù)的重要性數(shù)值型特征中位數(shù)/分組中位數(shù)對異常值魯棒注意方差壓縮問題類別型特征眾數(shù)/新增“未知”類別保留類別信息“未知”類別本身可能有預(yù)測力時序數(shù)據(jù)前值填充/插值保持時間連續(xù)性注意不要跨用戶填充金額/關(guān)鍵指標(biāo)刪除或人工核查填錯代價(jià)高保留缺失標(biāo)記供追溯5. 那些文檔里不會寫的踩坑記錄5.1 fillna的類型陷阱為什么我的字符串列沒被填充這是我在實(shí)際項(xiàng)目中最常遇到的問題之一。Scala API中df.na.fill(0)只會填充數(shù)值類型的列字符串列的null完全不受影響。更隱蔽的是如果你的DataFrame中某列是StringType但存儲的是數(shù)字字符串比如從CSV讀進(jìn)來沒做類型轉(zhuǎn)換fill(0)也不會生效。// 假設(shè)df有name(String)和age(Integer)兩列 // 這個操作只會填充age列的nullname列的null原封不動 df.na.fill(0) // 正確的做法用Map明確指定每列的填充值 df.na.fill(Map(name - unknown, age - 0))在PySpark中fillna(0)的行為是嘗試填充所有數(shù)值列但如果你傳入的是字符串它只會填充字符串列。這個差異在Scala和PySpark之間切換時特別容易搞混。實(shí)操建議無論用哪個語言始終用字典/Map形式指定填充值不要依賴默認(rèn)的類型匹配行為。5.2 dropna的subset參數(shù)你以為刪的是全部其實(shí)只看了部分列dropna的subset參數(shù)如果不指定默認(rèn)會考慮所有列。這意味著只要任意一列有null整行就會被刪除。在一個有50列的表里即使每列只有1%的缺失率一行完全沒有任何缺失的概率也只有(0.99)^50 ≈ 60%也就是說40%的行會被刪掉。# 危險(xiǎn)操作不指定subset可能刪掉大量數(shù)據(jù) df.dropna() # 等價(jià)于 df.dropna(howany, subsetNone) # 安全操作只考慮關(guān)鍵列 df.dropna(subset[user_id, order_id, amount])我的經(jīng)驗(yàn)是永遠(yuǎn)不要在不指定subset的情況下調(diào)用dropna。先想清楚哪些列是“關(guān)鍵列”——這些列的缺失意味著這行數(shù)據(jù)不可用其他列的缺失可以通過填充解決。5.3 用0填充的隱藏風(fēng)險(xiǎn)均值和方差的雙重扭曲用0填充數(shù)值型缺失值是最簡單粗暴的做法但它的副作用比大多數(shù)人想象的要大。假設(shè)一個字段的真實(shí)均值是100標(biāo)準(zhǔn)差是20你有20%的數(shù)據(jù)缺失并用0填充那么填充后的均值會變成80標(biāo)準(zhǔn)差會膨脹到約45。這意味著任何基于均值的統(tǒng)計(jì)都會偏低任何基于方差的分析都會偏高如果后續(xù)做標(biāo)準(zhǔn)化z-score填充的0會被當(dāng)成一個極端的低值# 對比用0填充 vs 用中位數(shù)填充對方差的影響 from pyspark.sql.functions import mean, stddev # 原始數(shù)據(jù)無缺失 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() # 用中位數(shù)填充 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()實(shí)測下來用中位數(shù)填充對方差的影響遠(yuǎn)小于用0填充。如果業(yè)務(wù)上確實(shí)需要用0填充比如“購買次數(shù)”字段那至少要在后續(xù)分析中加一個“是否填充”的標(biāo)記列讓模型或分析師知道哪些值是填充出來的。5.4 分布式環(huán)境下的approxQuantile精度問題approxQuantile是一個近似算法它的結(jié)果依賴于數(shù)據(jù)的分區(qū)方式。在分布式環(huán)境中如果數(shù)據(jù)傾斜嚴(yán)重某些分區(qū)的數(shù)據(jù)量遠(yuǎn)大于其他分區(qū)近似結(jié)果的誤差可能會超過你設(shè)定的relative error。# 檢查分區(qū)是否均勻 df.rdd.getNumPartitions() df.groupBy(spark_partition_id()).count().show() # 如果數(shù)據(jù)傾斜先repartition再計(jì)算分位數(shù) df_repartitioned df.repartition(200, region) median_val df_repartitioned.approxQuantile(salary, [0.5], 0.01)[0]另外approxQuantile對null值的處理也需要注意——它會自動忽略null值所以計(jì)算出的中位數(shù)是基于非null數(shù)據(jù)的。這通常是符合預(yù)期的但如果你需要把null也考慮進(jìn)去比如把null當(dāng)成0那就要先填充再計(jì)算。5.5 多列聯(lián)合填充時的順序依賴問題當(dāng)你需要對多列進(jìn)行填充且填充邏輯之間存在依賴關(guān)系時操作的順序會影響最終結(jié)果。比如你先用“年齡”的中位數(shù)填充了年齡缺失然后又想用“年齡”和“收入”的關(guān)系來填充收入缺失——這時候填充后的年齡已經(jīng)不再是原始數(shù)據(jù)了基于它計(jì)算出的收入填充值也會有偏差。# 有依賴關(guān)系的填充順序很重要 # 第一步先填充基礎(chǔ)字段 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)) )我的建議是盡量讓每列的填充邏輯獨(dú)立如果確實(shí)有依賴關(guān)系在代碼中明確注釋依賴順序并且對每一步的結(jié)果做數(shù)據(jù)質(zhì)量檢查。6. 把缺失值處理嵌入完整的ETL流水線6.1 設(shè)計(jì)可復(fù)用的缺失值處理函數(shù)在實(shí)際項(xiàng)目中缺失值處理不應(yīng)該是散落在各個腳本里的零散代碼而應(yīng)該封裝成可復(fù)用的函數(shù)或類。這樣做的價(jià)值在于統(tǒng)一處理標(biāo)準(zhǔn)、方便審計(jì)和修改、減少重復(fù)代碼。def clean_missing_values(df, config): 通用的缺失值處理函數(shù) config: { drop_cols: [col1, col2], # 缺失超過閾值直接刪除的列 fill_zero: [count1, count2], # 用0填充的列 fill_median: [salary, age], # 用中位數(shù)填充的列 fill_mode: [city, category], # 用眾數(shù)填充的列 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. 刪除關(guān)鍵列缺失的行 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. 用中位數(shù)填充 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. 用眾數(shù)填充 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這個函數(shù)的設(shè)計(jì)思路是把“策略”和“執(zhí)行”分離。配置文件描述每列該怎么處理函數(shù)負(fù)責(zé)執(zhí)行。這樣當(dāng)業(yè)務(wù)規(guī)則變化時只需要改配置不需要改代碼。6.2 處理前后的數(shù)據(jù)質(zhì)量校驗(yàn)缺失值處理做完之后必須做一輪校驗(yàn)確認(rèn)處理結(jié)果符合預(yù)期。校驗(yàn)的內(nèi)容包括目標(biāo)列的缺失率是否降到了預(yù)期水平填充后的數(shù)據(jù)分布是否發(fā)生了劇烈變化均值、方差、分位數(shù)是否有意外的類型轉(zhuǎn)換比如整數(shù)列被填充后變成了浮點(diǎn)列def validate_cleaning(df_before, df_after, target_cols): 對比處理前后的數(shù)據(jù)質(zhì)量 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 標(biāo)準(zhǔn)差: {before_stats[std]:.2f} - {after_stats[std]:.2f})如果發(fā)現(xiàn)填充后均值偏移超過10%或者標(biāo)準(zhǔn)差變化超過20%就需要重新審視填充策略是否合理。6.3 增量數(shù)據(jù)處理中的缺失值策略一致性在流式或增量ETL場景中缺失值處理有一個額外的挑戰(zhàn)今天用中位數(shù)填充明天數(shù)據(jù)更新后中位數(shù)變了歷史數(shù)據(jù)的填充值和新數(shù)據(jù)的填充值不一致。這會導(dǎo)致同一列的數(shù)據(jù)在不同批次之間不可比。解決方法是把填充參數(shù)持久化。每次計(jì)算出的中位數(shù)、眾數(shù)等統(tǒng)計(jì)量保存到配置表或文件中后續(xù)批次復(fù)用同一套參數(shù)定期比如每月重新計(jì)算并更新。# 保存填充參數(shù) 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) # 后續(xù)批次讀取參數(shù) with open(/path/to/fill_params.json, r) as f: params json.load(f) df df.fillna({salary: params[salary_median]})這個做法看起來簡單但在實(shí)際項(xiàng)目中經(jīng)常被忽略導(dǎo)致數(shù)據(jù)口徑不一致的問題很難排查。7. 幾個讓我印象深刻的真實(shí)案例7.1 一個字段的缺失率從5%突然漲到40%某次日常巡檢發(fā)現(xiàn)用戶表的“手機(jī)號”字段缺失率從平時的5%突然漲到了40%。第一反應(yīng)是上游數(shù)據(jù)同步出了問題但檢查了同步任務(wù)日志沒有發(fā)現(xiàn)異常。后來做了一次分組統(tǒng)計(jì)發(fā)現(xiàn)缺失集中在某個特定渠道的新增用戶上。進(jìn)一步排查發(fā)現(xiàn)那個渠道的注冊頁面最近改版了手機(jī)號從必填變成了選填。這不是數(shù)據(jù)質(zhì)量問題而是業(yè)務(wù)規(guī)則變化。如果當(dāng)時直接按“缺失率超過30%就刪除整列”的規(guī)則處理就會把整個手機(jī)號字段刪掉影響后續(xù)的短信觸達(dá)和用戶召回。這個案例給我的教訓(xùn)是缺失率的變化趨勢比缺失率的絕對值更重要。建立缺失率的日常監(jiān)控比事后補(bǔ)救有效得多。7.2 用均值填充導(dǎo)致的報(bào)表偏差一個銷售報(bào)表中“訂單金額”字段有約8%的缺失。當(dāng)時的處理邏輯是用均值填充。上線一個月后業(yè)務(wù)方反饋報(bào)表中的“總銷售額”比財(cái)務(wù)系統(tǒng)的數(shù)字高了約3%。排查后發(fā)現(xiàn)缺失的訂單金額并不是隨機(jī)分布的——它們主要來自一個大客戶這個客戶的訂單金額普遍偏高。用全局均值填充后這些訂單的金額被低估了但同時又因?yàn)樘畛淞酥刀鴽]有被排除在統(tǒng)計(jì)之外導(dǎo)致總銷售額出現(xiàn)了偏差。正確的做法應(yīng)該是對于這種非隨機(jī)缺失要么刪除缺失行并在報(bào)表中注明“不含缺失數(shù)據(jù)”要么用該客戶的歷史平均訂單金額來填充而不是用全局均值。7.3 時序填充時跨用戶污染的問題在處理用戶日活數(shù)據(jù)時需要補(bǔ)齊每個用戶每天的記錄。最初的實(shí)現(xiàn)是用last_value按時間排序填充但忘記加partitionBy(user_id)導(dǎo)致用戶A的缺失值被用戶B的數(shù)據(jù)填充了。這個bug在測試環(huán)境沒有發(fā)現(xiàn)因?yàn)闇y試數(shù)據(jù)只有少量用戶排序后恰好沒有跨用戶的情況。修復(fù)方法很簡單加上partitionBy即可。但這個案例說明時序填充一定要確認(rèn)窗口函數(shù)的partitionBy是否正確否則填充結(jié)果完全是錯的。# 錯誤寫法沒有按用戶分區(qū) window_wrong Window.orderBy(date) \ .rowsBetween(Window.unboundedPreceding, Window.currentRow) # 正確寫法按用戶分區(qū) window_correct Window.partitionBy(user_id).orderBy(date) \ .rowsBetween(Window.unboundedPreceding, Window.currentRow)8. 關(guān)于缺失值處理我總結(jié)的幾條實(shí)操原則做了這么多年的數(shù)據(jù)清洗關(guān)于缺失值處理我形成了幾個固定的判斷原則分享出來供參考。原則一先畫像再動手。不要拿到數(shù)據(jù)就寫fillna或dropna。先花10分鐘做一輪缺失值統(tǒng)計(jì)搞清楚每列的缺失比例、缺失的分布特征、缺失與其他列的關(guān)系。這10分鐘的投入能幫你避免后面10個小時的返工。原則二能保留標(biāo)記就保留標(biāo)記。無論你最終選擇刪除還是填充都建議在DataFrame中增加一個布爾列標(biāo)記原始數(shù)據(jù)是否缺失。這個標(biāo)記列本身可能就是一個有價(jià)值的特征而且在后續(xù)排查問題時可以快速定位哪些數(shù)據(jù)是填充出來的。原則三填充策略要能解釋給業(yè)務(wù)方聽。如果你用中位數(shù)填充了“收入”字段業(yè)務(wù)方問起來你要能說清楚為什么用中位數(shù)而不是均值、為什么不用0、填充后的數(shù)據(jù)對分析結(jié)論有什么影響。如果解釋不清楚說明填充策略可能有問題。原則四處理邏輯要版本化。缺失值處理的配置哪些列刪、哪些列填、填什么值應(yīng)該像代碼一樣納入版本管理。每次修改都要有記錄方便回溯和審計(jì)。原則五不要追求零缺失。有些缺失是數(shù)據(jù)本身的特性強(qiáng)行填充反而會引入噪聲。比如用戶畫像中的“年收入”字段很多人就是不愿意填你硬要填一個預(yù)測值可能還不如保留null讓模型自己去處理。最后分享一個我在實(shí)際工作中常用的小技巧對于不確定該填什么值的列我會先用不同的策略各跑一遍填0、填中位數(shù)、填眾數(shù)、刪除然后對比每種策略下下游核心指標(biāo)的變化。如果某種策略導(dǎo)致核心指標(biāo)偏移超過5%就說明這個策略可能不合適。這個方法雖然笨但很有效尤其是在面對不熟悉的業(yè)務(wù)數(shù)據(jù)時。