如何通过循环将分组聚合结果追加至空DataFrame？

问题分析 #

你当前的循环会为每个分组列生成独立的DataFrame，但直接尝试追加到空DataFrame（或转字符串追加）会因为Spark DataFrame的不可变性和结构不统一导致失败：

• Spark DataFrame是不可变对象，不能直接在循环中修改赋值；
• 每个聚合后的DataFrame的第一列是当前分组列的名称（比如x是"colA"，列名就是"colA"），列名不统一无法直接合并。

解决方案 #

我们需要先统一所有聚合DataFrame的结构，再批量合并，同时优化性能（避免重复计算总条数）：

1. 导入依赖函数 #

import org.apache.spark.sql.functions.{countDistinct, lit}

2. 预计算总条数 #

循环中重复计算df.count()会严重影响性能，先一次性计算并存储：

val totalCount = df.count()

3. 生成统一结构的聚合DataFrame列表 #

为每个分组列生成聚合结果，同时统一列名并添加标识列（记录当前是哪个分组列的结果）：

val aggregatedDfs = groupcols.map { colName =>
  df.groupBy(colName)
    .agg(
      countDistinct("ID").alias("ID_Count"),
      (countDistinct("ID") / totalCount).alias("Percentage")
    )
    // 将原分组列重命名为统一的"Group_Value"
    .withColumnRenamed(colName, "Group_Value")
    // 添加列记录原分组列的名称
    .withColumn("Group_Column", lit(colName))
}

4. 合并所有DataFrame #

通过unionByName合并列表中所有结构一致的DataFrame：

val finalDf = if (aggregatedDfs.nonEmpty) {
  aggregatedDfs.reduce(_ unionByName _)
} else {
  // 处理groupcols为空的情况，生成结构匹配的空DataFrame
  val emptySchema = aggregatedDfs.head.schema
  df.sparkSession.createDataFrame(df.sparkSession.emptyRDD, emptySchema)
}

5. 查看最终结果 #

finalDf.show()

关键说明 #

• 使用unionByName而非union：确保列名匹配时合并，避免因列顺序不一致导致错误；
• 统一结构：每个聚合DataFrame都包含Group_Column（原分组列名）、Group_Value（分组值）、ID_Count、Percentage四列，保证合并后数据可清晰溯源；
• 性能优化：预计算总条数避免循环内重复触发Action操作。

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