Spark Scala使用agg聚合时无法解析next_intent_temp列问题

你的代码存在两处关键语法错误，这导致Spark无法正确解析逻辑计划——尽管mydf.show()能正常显示列，但后续聚合操作的语法结构被破坏，最终引发列找不到的错误：

错误点1：withColumn调用未闭合 #

mydf定义中，when表达式的外层缺少闭合括号，导致withColumn方法未正确结束：

// 错误写法
val mydf = df.withColumn("next_intent_temp", when (col("next_intent")=== "val", 1).otherwise(0)
// 正确写法，补全右括号
val mydf = df.withColumn("next_intent_temp", when(col("next_intent") === "val", 1).otherwise(0))

错误点2：groupBy调用未闭合 #

groupBy方法的参数列表缺少闭合括号，导致后续的.agg被错误解析为groupBy的参数一部分，Spark无法识别next_intent_temp属于mydf的列：

// 错误写法
val mynewdf = mydf.groupBy(col("a"), col("b"), col("c")
              .agg(sum(col("next_intent_temp")).as("next_intent"))
// 正确写法，补全右括号
val mynewdf = mydf.groupBy(col("a"), col("b"), col("c"))
              .agg(sum(col("next_intent_temp")).as("next_intent"))

额外注意点 #

你的select操作中引用了market列，但该列既不在groupBy的分组列中，也未被聚合函数包裹，这会触发Spark分组查询规则错误，后续需要确保select中的列要么是分组键，要么是聚合结果。

修正后的完整代码示例：

def process(spark: SparkSession, df : DataFrame): DataFrame= {
  import spark.implicits._
  import org.apache.spark.sql.functions._

  val mydf = df.withColumn("next_intent_temp", when(col("next_intent") === "val", 1).otherwise(0))
  
  val mynewdf = mydf.groupBy(col("a"), col("b"), col("c"))
                .agg(sum(col("next_intent_temp")).as("next_intent"))
                // 假设将分组列a映射为market，根据实际需求调整
                .select(col("a").as("market"), struct(col("next_intent")).alias("data"))

  mynewdf
}

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