Spark Scala：嵌套DataFrame转Dataset遇AnalysisException报错求助

你遇到的错误本质是聚合后的DataFrame列结构和case class的嵌套结构不匹配：

• 聚合操作后，你的inputFlowRecordsAgg里的列都是扁平化的带前缀字段（比如FlowI.key、FlowS.minFlowTime）；
• 但InputFlowV1 case class要求的是顶级的嵌套结构体（一个FlowI对象和一个FlowS对象），Spark无法自动将扁平化的前缀列映射为嵌套结构体，因此报错找不到FlowI这个顶级列。

你需要将聚合后的扁平化字段重新打包成FlowI和FlowS两个嵌套结构体，再转换为目标Dataset。具体步骤如下：

1. 使用Spark的struct()函数，将所有FlowI.xxx字段打包成一个名为FlowI的结构体列；
2. 同样用struct()函数，将所有FlowS.xxx字段打包成名为FlowS的结构体列；
3. 只保留这两个新的嵌套列，再调用as[InputFlowV1]完成转换。

import org.apache.spark.sql.functions.{struct, min, max, sum, first, col}

def getReducedFlowR(inputFlowRecords: Dataset[InputFlowV1], @transient spark: SparkSession): Dataset[InputFlowV1]={
  val inputFlowRecordsAgg = inputFlowRecords.groupBy(column("FlowI.key") as "FlowI.key")
    .agg(
      min("FlowS.minFlowTime") as "FlowS.minFlowTime",
      max("FlowS.maxFlowTime") as "FlowS.maxFlowTime",
      sum("FlowS.flowStartedCount") as "FlowS.flowStartedCount",
      first("FlowI.Mac") as "FlowI.Mac",
      first("FlowI.SrcIP") as "FlowI.SrcIP",
      first("FlowI.DestIP") as "FlowI.DestIP",
      first("FlowI.DestPort") as "FlowI.DestPort",
      first("FlowI.L4Protocol") as "FlowI.L4Protocol",
      first("FlowI.Direction") as "FlowI.Direction",
      first("FlowI.Status") as "FlowI.Status"
    )

  // 组装嵌套结构体，字段顺序需与case class定义完全匹配
  val nestedDF = inputFlowRecordsAgg.select(
    struct(
      col("FlowI.Mac"),
      col("FlowI.SrcIP"),
      col("FlowI.DestIP"),
      col("FlowI.DestPort"),
      col("FlowI.L4Protocol"),
      col("FlowI.Direction"),
      col("FlowI.Status"),
      col("FlowI.key")
    ).alias("FlowI"),
    struct(
      col("FlowS.minFlowTime"),
      col("FlowS.maxFlowTime"),
      col("FlowS.flowStartedCount")
    ).alias("FlowS")
  )

  nestedDF.printSchema() // 此时schema会与InputFlowV1的结构完全匹配
  return nestedDF.as[InputFlowV1]
}

• 务必保证struct()中字段的顺序和case class的定义完全一致（比如FlowI的case class字段顺序是Mac, SrcIP, DestIP, DestPort, L4Protocol, Direction, Status, key，所以struct里也要严格遵循这个顺序）；
• 如果IPAddress是自定义类型，你需要确保Spark已经注册了对应的编码器（Encoder），否则可能需要在select阶段添加自定义逻辑，将SrcIP.bytes和DestIP.bytes转换为IPAddress对象。

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