做數(shù)據(jù)的人大概都遇到過這種場景業(yè)務(wù)方丟過來一張表字段是按月份展開的什么202401_sales、202402_sales、202403_sales一路排下去又或者是一行記錄里同時塞了十幾個指標列。你要做統(tǒng)計分析、要出報表第一步就是把這些“橫著”的列變成“豎著”的行。這個動作在行業(yè)里叫列轉(zhuǎn)行也叫寬表轉(zhuǎn)長表、unpivot。在 Spark DataFrame 里實現(xiàn)列轉(zhuǎn)行常規(guī)的做法有stack函數(shù)、explode函數(shù)還有一堆人繞遠路的union方案。這篇文章我就結(jié)合自己在某項目里的真實處理過程把 Spark DataFrame 列轉(zhuǎn)行的幾種寫法、選型邏輯、參數(shù)坑、性能調(diào)優(yōu)一次說清楚。先說結(jié)論如果只是普通的多列轉(zhuǎn)多行stack是首選如果你處理的是 Array 或 Map 類型的嵌套數(shù)據(jù)explode才派得上用場其他場景下用union硬懟基本都是自找麻煩。為什么這么說下面拆開講。1. 項目場景與需求拆解1.1 接到一個什么需求事情是這樣的。某個數(shù)據(jù)平臺項目里上游給我們同步了一張渠道推廣效果表表結(jié)構(gòu)大致是這樣每一行是一個渠道渠道名和渠道 ID 是固定的維度字段然后后面跟著jan_amount、feb_amount、mar_amount這種十二個月份的轉(zhuǎn)化金額列再加若干分組字段。表本身不算大幾千萬行的規(guī)模但如果直接拿這張寬表去做同比環(huán)比、趨勢分析、前端圖表渲染會非常別扭——因為按月分析本質(zhì)上需要一個“月份”維度而寬表里月份是散在列名里的。當時數(shù)據(jù)組拿到的需求原文很明確把這張寬表里每個月的數(shù)據(jù)拆成獨立一行輸出channel_id、channel_name、month、amount四個核心字段方便下游直接掛 BI 出趨勢圖。這種“列名即值”的表結(jié)構(gòu)就是列轉(zhuǎn)行最典型的適用場景。1.2 列轉(zhuǎn)行的本質(zhì)是什么列轉(zhuǎn)行聽起來是個高大上的數(shù)據(jù)操作其實底層邏輯一句話就能講透把多個列的值連同它們對應(yīng)的列名重新組織成“行記錄”中的字段。你可以理解成把一張 A4 紙橫著折一下原本橫向排列的信息變成縱向羅列。在 Spark 的 DataFrame / SQL 模型里列轉(zhuǎn)行本質(zhì)上是對“列名到值”這一組映射關(guān)系做展開。最常見的實現(xiàn)手段有兩個方向使用stack系列函數(shù)直接展開多列成多行使用explode系列函數(shù)拆解行內(nèi)的集合類型字段前者解決的是“結(jié)構(gòu)規(guī)整的寬表”后者解決的是“字段本身是數(shù)組或字典”的復雜結(jié)構(gòu)。兩者適用場景不同但很多人一搜列轉(zhuǎn)行就看見explode順手就用了結(jié)果發(fā)現(xiàn)根本不靈——因為普通列根本沒法直接 explode。這也是我在實際項目中反復跟同事強調(diào)的一點先看數(shù)據(jù)類型再選工具。1.3 技術(shù)選型前的幾個判斷維度在動手寫代碼之前我一般會先問自己三個問題。這看起來有點多余但實踐證明想清楚再寫比瞎寫半天再返工高效得多。第一列的數(shù)量多不多、是否固定。如果列是動態(tài)生成的比如這個月新增了一個字段那代碼里就不能硬編碼列名得考慮動態(tài)拼接。第二每個要轉(zhuǎn)置的列的數(shù)據(jù)類型是否一致。有的是 String有的是 Long有的是 Double如果不統(tǒng)一stack會直接報類型不匹配或者運行時拋異常。第三有沒有需要保留的維度列。列轉(zhuǎn)行之后原來一行里的維度字段比如渠道 ID需要復制到每一行里這個行為stack天然支持但很多人寫的時候容易漏。這三個問題想清楚之后代碼其實就是一個模板的事。但是模板背后的參數(shù)和坑才是真正值得寫下來的部分。2. 核心方案解析為什么 stack 是第一選擇2.1 stack 函數(shù)的工作原理stack是 Hive / Spark SQL 里的一個 UDTF用戶定義表生成函數(shù)作用是把 N 列數(shù)據(jù)展開成 M 行其中 N 是參與展開的列數(shù)M 由你指定的行數(shù)決定。它的標準語法長這樣stack(n, col1, col2, ..., colk)這里的n表示一共要拆成多少行拆完之后每一行有幾列取決于你后續(xù)AS子句里定義的字段數(shù)量。舉個最直白的例子如果你寫stack(2, col_a, col_b, col_c, col_d)意思就是把這四個列的值拆成 2 行每行包含 2 列值。第 1 行的兩列取自col_a和col_b第 2 行的兩列取自col_c和col_d。初次接觸的人很容易繞暈的點就在這里stack的第一個參數(shù)是“輸出行數(shù)”而不是“輸入列數(shù)”。輸入的列數(shù)必須是n的整數(shù)倍多出來的列會被按順序分到下一行。如果列數(shù)對不上加載階段報錯如果類型對不上運行階段報錯。在 DataFrame 里使用stack通常不會直接寫.select(col1, ...)這種普通選擇而是要用selectExpr或者把整個表達式當作一個字符串丟進去df.selectExpr( channel_id, channel_name, stack(12, jan_amount, feb_amount, mar_amount, apr_amount, may_amount, jun_amount, jul_amount, aug_amount, sep_amount, oct_amount, nov_amount, dec_amount) as (month_name, amount) )執(zhí)行完這個之后原來的 12 個金額列就會變成兩列輸出一列是月份名字段名month_name一列是金額字段名amount。整張 DataFrame 的行數(shù)會從原來的 1 行變成 12 行如果原來有 N 行就是 12N 行。2.2 參數(shù)細節(jié)和列名映射規(guī)則寫 stack 的時候有四個細節(jié)值得單獨提出來說全是踩過的坑換來的經(jīng)驗。第一個細節(jié)stack里列的順序決定了行的順序。上例中(month_name, amount)是兩列一組所以偶數(shù)位列名會落在month_name奇數(shù)位值會落在amount。如果你把列順序?qū)懛戳溯敵龅?value 和 key 就全錯位了。當時我們項目里就出現(xiàn)過一版數(shù)據(jù)月份和金額完全對不上排查了半天才發(fā)現(xiàn)是列順序?qū)懛?。第二個細節(jié)AS子句里的字段數(shù)量必須和stack展開后每一行的字段數(shù)量一致。如果你寫stack(6, col1, col2)每個輸出行只有 1 個字段那么AS (value)里就只能寫一個字段名寫兩個就報錯。這是一個非常容易犯的低級錯誤。第三個細節(jié)stack展開過程中如果某個原始值為null它不會自動幫你過濾掉而是會把 null 值照常輸出到目標字段。如果你的業(yè)務(wù)需要“只有有效數(shù)據(jù)的行”那你得在 stack 之后手動加過濾條件。這個坑當初讓我們多跑了一遍全量任務(wù)。第四個細節(jié)參與 stack 的列必須顯式寫出完整列表。不能寫stack(12, *)這種語法也不支持直接放一個列集合變量進去。對于幾十上百個字段的寬表手寫不太現(xiàn)實通常用代碼動態(tài)生成表達式字符串這個在下面第 3 部分會給出完整例子。2.3 為什么不用 union 方案我見過不少同事在沒接觸過 stack 的時候會用union加多次select去拼。思路大概是先 select 維度列 第一個月字段union 上 select 維度列 第二個月字段再 union 第三個月……這個方案不是不能跑但是它在代碼層面有非常明顯的毛病第一代碼量爆炸12 個月就要寫 12 段 select如果字段有幾十個腳本能寫到上千行第二每寫一個 union 子句Spark 都會多一層邏輯計劃節(jié)點解析、優(yōu)化、執(zhí)行的開銷都會增加第三可維護性極差哪天業(yè)務(wù)加了一個新字段你要從第 1 段改到最后一段漏改一個就很難發(fā)現(xiàn)。相比之下stack 用一個表達式的線性復雜度就搞定了邏輯清晰、代碼量小、性能也更好。除非你是在做某些極端復雜的轉(zhuǎn)換或者你的目標是把不同來源的數(shù)據(jù)按行拼接這種應(yīng)該用 union 而不是 stack否則在“寬表轉(zhuǎn)長表”這個場景下stack 的優(yōu)先級是最高的。3. 實操過程從數(shù)據(jù)模擬到完整實現(xiàn)3.1 構(gòu)造模擬數(shù)據(jù)為了把整個過程講透我模擬一份脫敏后的數(shù)據(jù)。假設(shè)我們有 3 個渠道每個渠道對應(yīng) 3 個月的銷售額為了演示方便只取 3 個月而不是 12 個月邏輯完全一樣from pyspark.sql import SparkSession spark SparkSession.builder.master(local).appName(column_to_row).getOrCreate() data [ (C001, 渠道A, 100, 150, 200), (C002, 渠道B, 300, 250, 280), (C003, 渠道C, 400, 420, 380), ] df spark.createDataFrame(data, [channel_id, channel_name, jan_amount, feb_amount, mar_amount]) df.show()輸出結(jié)果---------------------------------------------------- |channel_id|channel_name|jan_amount|feb_amount|mar_amount| ---------------------------------------------------- | C001| 渠道A| 100| 150| 200| | C002| 渠道B| 300| 250| 280| | C003| 渠道C| 400| 420| 380| ----------------------------------------------------這就是一個標準的待轉(zhuǎn)置寬表維度列是channel_id和channel_name需要展開的列是jan_amount、feb_amount、mar_amount。3.2 靜態(tài)寫法固定三列轉(zhuǎn)行最直接的方法是寫死三個列名用selectExpr完成轉(zhuǎn)換result_df df.selectExpr( channel_id, channel_name, stack(3, jan_amount, feb_amount, mar_amount) as (month, amount) ) result_df.show()執(zhí)行結(jié)果--------------------------------- |channel_id|channel_name|month|amount| --------------------------------- | C001| 渠道A| 100| ? ---------------------------------等一下這里就有個問題。stack(3, jan_amount, feb_amount, mar_amount)是把 3 個列拆成 3 行每行只有 1 列。所以AS (month, amount)的寫法是錯的——month拿到的其實是第一個輸入列的值也就是 100amount根本沒有對應(yīng)的分配系統(tǒng)會報錯。正確的寫法是讓每個輸出行有 2 個字段所以需要傳入 6 個值result_df df.selectExpr( channel_id, channel_name, stack(3, jan, jan_amount, feb, feb_amount, mar, mar_amount) as (month, amount) ) result_df.show()這次的結(jié)果才是對的--------------------------------- |channel_id|channel_name|month|amount| --------------------------------- | C001| 渠道A| jan| 100| | C001| 渠道A| feb| 150| | C001| 渠道A| mar| 200| | C002| 渠道B| jan| 300| | C002| 渠道B| feb| 250| | C002| 渠道B| mar| 280| | C003| 渠道C| jan| 400| | C003| 渠道C| feb| 420| | C003| 渠道C| mar| 380| ---------------------------------注意我在傳入序列里加了字符串常量jan、feb、mar。這是 stack 列轉(zhuǎn)行的核心套路你要輸出的“行標識字段”這里就是月份名其實就是列名本身但列名是靜態(tài)的文本不會跟著值走。所以你需要手動把這些常量對應(yīng)地放進參數(shù)序列里讓它們和真實列值一一配對。如果你希望輸出的月份是數(shù)字1、2、3就寫成result_df df.selectExpr( channel_id, channel_name, stack(3, 1, jan_amount, 2, feb_amount, 3, mar_amount) as (month, amount) )原理完全一樣只是常量從字符串變成了整數(shù)。3.3 動態(tài)寫法字段特別多怎么辦如果表里有幾十甚至上百個要轉(zhuǎn)置的列手動寫col1, col1, col2, col2, ...當然是不現(xiàn)實的。這時候就需要 Python / Scala 代碼來拼接表達式。以 Python PySpark 為例假設(shè)你的寬表字段名遵循xxx_amount這樣的規(guī)則可以先篩選出所有目標列名然后動態(tài)構(gòu)造 stack 表達式的參數(shù)列表amount_cols [c for c in df.columns if c.endswith(_amount)] # 按列名字典序排序保證結(jié)果穩(wěn)定 amount_cols.sort() # 構(gòu)造 stack 參數(shù)序列每個列前面加上它自己的名字文本 stack_params [] for c in amount_cols: month_name c.replace(_amount, ) stack_params.append(f{month_name}) # 輸出行的月份字段 stack_params.append(c) # 輸出行的金額字段 stack_expr fstack({len(amount_cols)}, {, .join(stack_params)}) as (month, amount) result_df df.selectExpr( channel_id, channel_name, stack_expr )這樣寫完之后即使下個月上游多了一個apr_amount字段你只要重新從df.columns里篩選代碼一行都不用改。動態(tài)拼接表達式算是這類需求里比較推薦的進階寫法。但這里有個前提你的列名必須有規(guī)律。如果列名亂得毫無章法比如a1、b2、c3那就得老老實實維護一個字段映射表手工指定哪些列要展開、對應(yīng)的輸出標識文本是什么。映射表的方式雖然要維護但勝在可控、不容易出錯。3.4 explode 的適用場景處理 Array 和 Map上面說了 stack 是列轉(zhuǎn)行的主力但還有一個高頻場景是會用到 explode 的當數(shù)據(jù)本身以 Array 或 Map 存儲時。比如上游給了你一行數(shù)據(jù)里面一個字段是score_list存的是這個渠道每一天的分數(shù)是一個 Array[Double]你要把它拆成一行行記錄。這種場景用 stack 就得先把 Array 展開到多個列如果 Array 長度不定根本沒法展開而 explode 就是專門干這個的from pyspark.sql.functions import explode data [ (C001, 渠道A, [100, 150, 200]), (C002, 渠道B, [300, 250, 280]), ] df spark.createDataFrame(data, [channel_id, channel_name, score_list]) result_df df.select( channel_id, channel_name, explode(score_list).alias(score) ) result_df.show()輸出--------------------------- |channel_id|channel_name|score| --------------------------- | C001| 渠道A| 100| | C001| 渠道A| 150| | C001| 渠道A| 200| | C002| 渠道B| 300| | C002| 渠道B| 250| | C002| 渠道B| 280| ---------------------------如果還需要知道當前拆出來的是這個數(shù)組里的第幾個元素用posexplode它會在輸出里額外帶一個索引列from pyspark.sql.functions import posexplode result_df df.select( channel_id, channel_name, posexplode(score_list).alias(pos, score) )posexplode在同時需要“序號”和“值”的場景里很實用比如你要把數(shù)組里的第一個值當作基準值那pos就派上用場了。至于是不是應(yīng)該先把多列合并成struct再用explode拆這種事情純粹看個人習慣。如果有人喜歡把 3 個月的金額先構(gòu)造成一個array再explode那也能達到列轉(zhuǎn)行的效果但多了一步構(gòu)造數(shù)組的開銷性能上不如直接用 stack 來得干脆。我的建議是普通的多列轉(zhuǎn)多行別繞路stack 直接上數(shù)組/字典里的嵌套數(shù)據(jù)才輪到 explode 發(fā)揮。3.5 類型不一致時的處理辦法stack 特別容易踩的坑之一是參與展開的列類型不統(tǒng)一。比如某個月的金額字段是 Double另外某個月由于上游數(shù)據(jù)問題被定義成了 String此時 Spark 在生成邏輯計劃時會發(fā)現(xiàn) stack 的輸入類型不一致直接拋 AnalysisException。解決辦法是在傳入 stack 之前先對所有列做一次類型統(tǒng)一from pyspark.sql.functions import col # 假設(shè) df 里有多個列需要轉(zhuǎn)置先統(tǒng)一轉(zhuǎn)成 Double for c in amount_cols: df df.withColumn(c, col(c).cast(double))這種操作在數(shù)據(jù)治理不嚴格的業(yè)務(wù)場景里非常常見。同一個業(yè)務(wù)字段因為上游報表口徑調(diào)整從 Decimal 變成 String再到 Double你根本不知道哪一天會變。所以在 stack 前統(tǒng)一 cast 是一個成本很低但收益很大的好習慣。4. 常見問題與排查技巧實錄4.1 報錯“The number of columns doesnt match”這個錯誤信息通常長這樣org.apache.spark.sql.AnalysisException: The number of columns produced by STACK should not be greater than number of columns in the query問題根源基本是stack里傳的列數(shù)和AS子句定義的字段數(shù)量對不上。舉個例子你寫了stack(3, jan, jan_amount, feb, feb_amount, mar, mar_amount)這 6 個值會被分成 3 行、每行 2 個字段所以AS 子句必須寫 2 個字段名。如果你寫成了as (month)就會報錯。排查思路很簡單數(shù)一下 stack 參數(shù)列表里一共有幾個值除以n得到的結(jié)果必須和 AS 里的字段數(shù)一致。4.2 類型不匹配的坑另一個高頻報錯是cannot resolve ... due to data type mismatch。參與 stack 的列類型五花八門有 String、有 Double、有 IntegerSpark 為了讓這些類型能放進同一個輸出字段需要找到一個公共類型找不到就直接報錯。解決方法前面已經(jīng)提過在 stack 前統(tǒng)一 cast。如果因為你懶不想寫 cast也可以用coalesce或者字符串拼接把它們統(tǒng)一成 String但這樣后續(xù)用起來還得再轉(zhuǎn)一次不太劃算。我個人建議是統(tǒng)一成業(yè)務(wù)最終使用的類型比如金額就統(tǒng)一成 Decimal(18, 2) 或 Double。4.3 合并 null 值的影響stack 不會主動過濾 null。如果你有 12 個月的金額列但某些月份確實沒有數(shù)據(jù)轉(zhuǎn)出來的 12 行里那些月份對應(yīng)的amount字段就是 null。下游聚合的時候 null 會被 sum 忽略但如果業(yè)務(wù)要的是“只保留有實際值的月份”你必須在 stack 之后加一個過濾條件result_df.filter(col(amount).isNotNull())這里有個容易忽略的細節(jié)如果你用coalesce把 null 替換成 0再去過濾就徹底失效了因為全變成 0 之后你根本分辨不出哪些是真實 0、哪些是缺失值。所以除非業(yè)務(wù)明確要求缺失按 0 計否則別輕易把 null 填成 0寧可多留一步讓下游自己決定怎么處理。4.4 執(zhí)行慢重點看這兩處列轉(zhuǎn)行本身不會引起數(shù)據(jù)膨脹因為原本一行里的數(shù)據(jù)拆成多行后值還是那些值只是形態(tài)變了。所以正常情況下這個操作不會比普通 select 慢很多。如果你發(fā)現(xiàn)任務(wù)運行特別慢排查重點應(yīng)該放在以下兩處第一檢查是不是有大字段或者數(shù)量巨大的列全部參與 stack。比如一張表有 200 個字段、100 個都要轉(zhuǎn)那么單條記錄的寬度極大序列化和網(wǎng)絡(luò)傳輸?shù)某杀径紩弦粋€臺階。建議先只 select 需要的字段再轉(zhuǎn)不要一上來就select *。第二檢查是不是在轉(zhuǎn)完之后立刻做了高代價的 join 或 groupBy。列轉(zhuǎn)行之后數(shù)據(jù)量不變但如果你轉(zhuǎn)完之后又和一張大表做 join那慢的根源其實是 join而不是列轉(zhuǎn)行本身。把執(zhí)行計劃拿過來看一下哪個 stage 耗時最長通常問題一目了然。4.5 一個實用小技巧動態(tài)生成 stack 時給字段加序號批量生成的字段比如多個月份列它們的順序在動態(tài)篩選后不一定可控。你可以主動給參數(shù)序列排序而不是依賴 DataFrame 的列順序。具體做法是把篩選出來的列名先排序再拼接表達式。這樣不管上游表字段怎么亂排你的輸出順序始終穩(wěn)定下游做對比測試也方便。另外一個容易被忽略的地方stack表達式里的字符串常量比如jan如果是變量拼 SQL 的時候記得轉(zhuǎn)義和加引號防止空格、特殊字符把表達式搞壞。我見過有人動態(tài)拼出的表達式里字符串寫成了month_2024-01結(jié)果日期里的連字符導致表達式被截斷排查確實費了些時間。穩(wěn)妥的做法是構(gòu)造表達式后用spark.sql跑一個select快速驗證或者打印出來看一遍再執(zhí)行。5. 項目落地中的一點個人經(jīng)驗列轉(zhuǎn)行這個操作看著簡單但在真實項目里它往往不是一次性跑完就結(jié)束的。數(shù)據(jù)入倉、數(shù)據(jù)清洗、ETL 調(diào)度、下游 BI 報表每一步都可能對輸出的字段命名、類型、null 值策略有不同要求。我在這個項目里做了一版標準化方案之后基本把它固化成了一個公共函數(shù)團隊里其他人遇到“寬表轉(zhuǎn)長表”就直接復用不再重復踩坑。公共函數(shù)的核心邏輯不復雜輸入 DataFrame 和一組“列名 → 輸出標識”的映射自動統(tǒng)一類型、動態(tài)拼 stack 表達式、輸出標準化字段。如果你們團隊經(jīng)常處理類似的報表數(shù)據(jù)我強烈建議一步到位做成通用工具省得每次手工改腳本。還有一點列轉(zhuǎn)行的結(jié)果最好直接落成 Parquet 格式的中間表字段命名要跟下游約定清楚。因為轉(zhuǎn)完之后原先藏在列名里的月份信息現(xiàn)在變成了一個字段的值這個值的數(shù)據(jù)質(zhì)量直接決定了下游過濾條件怎么寫。如果month字段有時間類型、字符串類型混著來下游查詢腳本就得寫兩套邏輯維護成本翻倍。最后說一個從實際操作里沉淀出來的心得column-to-row 的思路在很多 SQL 引擎里是通用的不只是 Spark。你如果熟悉了 Spark 的 stack后面接觸其他引擎里類似的 unpivot 函數(shù)會發(fā)現(xiàn)核心思想都是一樣的——把“元數(shù)據(jù)”從列名里解放出來變成真正的數(shù)據(jù)。這個理念在數(shù)據(jù)建模里也值得留意凡是長期要沿用的寬表多花一點時間把它轉(zhuǎn)成規(guī)整的長表后面的分析工作會輕松非常多。