Spark Scala多DataFrame关联时歧义列报错问题求助

原始DataFrame（rankedDF）数据 #

拆分DataFrame的代码 #

val numberRanks = rankedDF.select("rank").distinct().count().toInt

// 创建多个按rank拆分的DataFrame
val rankDFs = for (i <- 1 to numberRanks) yield {
  rankedDF.filter(col("rank") === i)
}

尝试的关联代码（报错） #

// 尝试关联多个DataFrame
val joinedDFs = rankDFs.reduce((df1, df2) => 
  df1.join(df2, Seq("job_id"))
  .withColumn("combination_ratings", array(col("rating"), col("rating")))
  .withColumn("combination_task", array(col("task_id"), col("task_id")))
  .withColumn("final_proba", col("proba") * col("proba"))
).select("job_id", "combination_task", "combination_ratings", "final_proba")

关联后预期中间数据 #

期望最终结果 #

报错信息 #

reference 'rating' is ambiguous, could be: rating, rating

解决方案 #

方案1：为拆分后的DataFrame重命名列（避免歧义） #

在拆分阶段给每个非job_id的列加上rank前缀，确保关联后列名唯一：

val numberRanks = rankedDF.select("rank").distinct().count().toInt

// 拆分时重命名列，规避后续关联冲突
val rankDFs = for (i <- 1 to numberRanks) yield {
  rankedDF.filter(col("rank") === i)
    .withColumnRenamed("task_id", s"task_id_$i")
    .withColumnRenamed("rating", s"rating_$i")
    .withColumnRenamed("proba", s"proba_$i")
    .drop("rank")
}

// 关联并构建目标结果
val joinedDFs = rankDFs.reduce((df1, df2) => df1.join(df2, Seq("job_id")))
  .withColumn("combination_task", array((1 to numberRanks).map(i => col(s"task_id_$i")):_*))
  .withColumn("combination_ratings", array((1 to numberRanks).map(i => col(s"rating_$i")):_*))
  .withColumn("final_proba", (1 to numberRanks).map(i => col(s"proba_$i")).reduce(_ * _))
  .select("job_id", "combination_task", "combination_ratings", "final_proba")

方案2：用分组聚合替代拆分+关联（更高效） #

无需拆分DataFrame，直接按job_id分组，结合窗口函数完成数组收集和乘积计算：

import org.apache.spark.sql.expressions.Window

// 按job_id分区、rank排序，保证数据顺序匹配rank
val windowSpec = Window.partitionBy("job_id").orderBy("rank")

val resultDF = rankedDF
  .withColumn("task_array", collect_list("task_id").over(windowSpec))
  .withColumn("rating_array", collect_list("rating").over(windowSpec))
  // Spark 3.0+可用product函数，低版本用exp(sum(log(proba)))替代
  .withColumn("proba_product", product("proba").over(windowSpec))
  // 保留每个job_id中包含所有rank数据的行
  .withColumn("max_rank", max("rank").over(Window.partitionBy("job_id")))
  .where(col("rank") === col("max_rank"))
  .select(
    "job_id",
    col("task_array").alias("combination_task"),
    col("rating_array").alias("combination_ratings"),
    col("proba_product").alias("final_proba")
  )

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