Spark Scala实现DataFrame按cnt_wk动态行展开 替代Union All方案咨询

针对你提到的宽表转长表需求，这里提供一套基于Spark内置函数的高效方案，完全避免手动过滤+Union All的低效操作，适配动态生成的wk1至wkn列，且能高效处理2亿+量级的数据。

核心思路 #

1. 动态识别wk列：从DataFrame schema中自动提取所有以wk开头的列，无需硬编码列名。
2. 数组化wk列：将所有wk列合并为一个数组列，方便后续按索引取值。
3. 生成并展开序列：用sequence函数生成从1到cnt_wk的整数序列，再通过explode将单行展开为多行。
4. 映射对应wk值：根据展开后的num_wks值，从数组中取出对应的wk列数据。

代码实现 #

假设你的源数据DataFrame名为sourceDF，包含id、cnt_wk以及动态的wk1、wk2...wkn列：

import org.apache.spark.sql.functions._

// 1. 动态提取所有wk列，确保按wk1、wk2...wkn的顺序排列
val wkColumns = sourceDF.columns.filter(_.startsWith("wk")).sorted

// 2. 将所有wk列合并为一个数组列
val dfWithWkArray = sourceDF.withColumn("wk_array", array(wkColumns.map(col): _*))

// 3. 生成1到cnt_wk的序列并展开为多行
val dfExploded = dfWithWkArray.withColumn(
  "num_wks",
  explode(sequence(lit(1), col("cnt_wk"))) // Spark 2.4+支持sequence函数
)

// 4. 根据num_wks取出对应的wk值，并保留需要的列
val resultDF = dfExploded.withColumn(
  "wk_value",
  element_at(col("wk_array"), col("num_wks")) // element_at支持1-based索引，正好匹配num_wks
).select("id", "cnt_wk", "num_wks", "wk_value")

关键说明 #

• 适配动态wk列：无需提前知道max(cnt_wk)，wkColumns会自动识别所有wk后缀的列，排序后保证数组顺序与wk1至wkn一致。
• 高效处理大数据：所有操作均使用Spark内置的分布式函数，避免了手动Union All带来的任务调度开销，适合2亿+数据量的场景。
• 边界情况处理：如果存在cnt_wk=0的行，sequence(1, 0)会生成空数组，explode后这类行将被自动过滤；若需保留，可提前添加过滤逻辑或调整序列生成规则。
• 版本兼容性：sequence和element_at函数从Spark 2.4版本开始支持，若使用更低版本，可替换为posexplode结合range的方式实现序列生成。

内容的提问来源于stack exchange，提问作者Appden65