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