Spark Scala实现层级结构数据表的扁平化转换

嘿，我来帮你解决这个层级数据扁平化的问题！先理清楚咱们的需求：你有一个存储地理层级（国家→省→城市→区县）的树状表，现在要把每条记录的完整层级链展开成单独的列，方便后续分析或展示。

先看咱们的输入和期望输出 #

输入表结构和数据是这样的：

期望的扁平化输出应该每条记录都包含从根节点到当前节点的所有层级的ID和名称，比如Brampton那行就要显示出它所属的城市、省份、国家信息，像这样：

实现思路 #

咱们的表已经有path字段了，这简直是福音！它直接存了当前节点到根节点的ID路径，咱们可以利用这个字段拆分出所有祖先ID，再关联原表拿到对应的名称，最后动态生成每个层级的列。如果没有path字段也没关系，后面我会给你补充递归CTE的方案。

完整的Spark Scala代码实现 #

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

// 第一步：先模拟输入数据（实际场景换成你自己的数据源，比如读Hive表、CSV等）
val inputData = Seq(
  (55, "Canada", null, null, 0),
  (77, "Ontario", 55, "/55", 1),
  (100, "Toronto", 77, "/55/77", 2),
  (104, "Brampton", 100, "/55/77/100", 3)
).toDF("Id", "name", "parentId", "path", "depth")

// 第二步：把path转成包含所有层级ID的数组（从根到当前节点）
val withFullPath = inputData.withColumn(
  "full_id_path",
  // 根节点的path是null，直接把自己的Id放进数组
  when(col("path").isNull, array(col("Id").cast(StringType)))
    // 非根节点：把path和当前Id拼接后拆分，过滤掉开头的空字符串（因为path以/开头）
    .otherwise(split(concat(col("path"), "/", col("Id").cast(StringType)), "/"))
    .filter(_ =!= "")
    .cast(ArrayType(IntegerType))
)

// 第三步：拿到最大的层级深度，这样就能动态生成对应数量的列，不用硬编码
val maxDepth = withFullPath.select(max(col("depth"))).head().getInt(0)

// 第四步：创建临时视图，方便后面关联查询每个Id对应的名称
inputData.createOrReplaceTempView("hierarchy_table")

// 第五步：循环生成每个层级的Id和Name列
val flattenedDF = (0 to maxDepth).foldLeft(withFullPath) { (df, level) =>
  // 先取出当前层级的Id
  val levelIdCol = s"Level${level}_Id"
  df.withColumn(levelIdCol, element_at(col("full_id_path"), level + 1))
    // 关联原表拿到对应的名称
    .join(
      inputData.select(col("Id").as(levelIdCol), col("name").as(s"Level${level}_Name")),
      levelIdCol,
      "left_outer" // 层级不够的节点用null填充
    )
}

// 第六步：选择最终需要的列，把原始列和所有层级列整合起来
val finalDF = flattenedDF.select(
  col("Id"), col("name"), col("parentId"), col("depth"),
  // 把每个层级的Id和Name列展开成序列
  (0 to maxDepth).flatMap(level => Seq(col(s"Level${level}_Id"), col(s"Level${level}_Name"))): _*
)

// 看看结果！
finalDF.show(false)

代码里的关键细节 #

• 处理path字段：根节点的path是null，所以单独处理；非根节点把path和当前Id拼接后拆分，过滤掉拆分后开头的空字符串（因为path是/55这种格式，拆分后第一个元素是空）。
• 动态生成列：用foldLeft循环遍历每个层级，避免硬编码层级数，这样哪怕你的数据有10层也能自动适配。
• 左关联填充null：层级不够的节点（比如根节点只有Level0），后面的层级列会用null填充，保证结构统一。

如果没有path字段怎么办？用递归CTE！ #

要是你的表没有path字段，咱们可以用Spark的递归CTE来构建完整的层级ID路径，代码如下：

// 先把输入表创建成临时视图
inputData.createOrReplaceTempView("hierarchy_table")

// 递归CTE构建层级路径
val recursiveDF = spark.sql("""
  WITH RECURSIVE hierarchy_cte AS (
    -- 锚点：根节点（parentId为null），full_id_path就是自己的Id
    SELECT Id, name, parentId, depth, array(Id) as full_id_path
    FROM hierarchy_table
    WHERE parentId IS NULL
    UNION ALL
    -- 递归：子节点关联父节点，把自己的Id加到父节点的full_id_path里
    SELECT h.Id, h.name, h.parentId, h.depth, array_union(c.full_id_path, array(h.Id)) as full_id_path
    FROM hierarchy_table h
    JOIN hierarchy_cte c ON h.parentId = c.Id
  )
  SELECT * FROM hierarchy_cte
""")

得到recursiveDF之后，后面的步骤和之前完全一样，基于full_id_path生成扁平化列就可以了。

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