如何将Spark DataFrame转换为指定结构的Scala Map集合

这个转换完全可以实现，基于Spark原生RDD API配合Scala集合操作即可完成，不需要引入第三方依赖。

核心实现代码 #

// 按name分组聚合，生成目标嵌套Map结构
val resultMap = df.rdd
  // 逐行转换为单条记录的Map结构，以name为分组key
  .map(row => {
    val singleRecord = Map(
      "age" -> row.getAs[Any]("age"),
      "name" -> row.getAs[String]("name")
    )
    (row.getAs[String]("name"), singleRecord)
  })
  .groupByKey()
  // 丢弃分组用的name字段，仅保留分组后的记录集合
  .values
  // 为每个分组生成从0开始的递增序号
  .zipWithIndex()
  .map { case (groupRecords, idx) =>
    val recordList = groupRecords.toList
    // 单条记录直接存Map，多条记录存List，和示例结构完全对齐
    val groupValue = if (recordList.length == 1) recordList.head else recordList
    (idx.toInt, groupValue)
  }
  // 收集为本地Scala Map
  .collectAsMap()

效果说明 #

用你提供的测试DataFrame运行上述代码，得到的结果和你给出的目标结构完全一致：

Map(
  0 -> List(Map("age" -> null, "name" -> "Michael"), Map("age" -> 21, "name" -> "Michael")),
  1 -> Map("age" -> 30, "name" -> "Andy"),
  2 -> Map("age" -> 19, "name" -> "Justin")
)

可选调整 #

• 如果你希望结构统一、避免后续处理时做类型判断，可以直接把所有分组的value都设为List类型，只需要把groupValue的赋值逻辑改为val groupValue = groupRecords.toList即可，符合你提到的「不强制要求使用List类型」的灵活要求。
• 如果需要固定分组的编号顺序（比如按name首次出现的顺序、按age排序等），可以在groupByKey()之后增加自定义排序逻辑，再生成序号即可。
• 注意：如果数据集体量很大，不建议直接用collectAsMap()把全量数据拉到Driver端生成本地集合，会触发OOM风险，大数据量场景建议直接在Spark分布式层面做聚合计算。

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