Databricks中Spark代码报类型不匹配错误，请求排查问题

问题原因 #

• getnearestFiveMinSlot函数定义时要求接收Int类型参数，但你在withColumn中传入的col("slotValue")是org.apache.spark.sql.Column类型，类型不匹配直接触发报错。
• 函数内部使用collect()属于Spark的Action操作，会将查询结果拉取到Driver节点，这种方式无法和DataFrame的分布式Column操作兼容，不能实现逐行处理数据的需求。

修复方案 #

方案一：用Spark内置函数实现（推荐，性能最优） #

你的需求本质是找到大于等于给定值的最小5分钟间隔（300秒的倍数），直接通过数学计算即可实现，无需临时表查询：

import org.apache.spark.sql.functions._

val slotValue = List(100,100,100,4,5)
val df = slotValue.toDF("slotValue")

// 计算大于等于slotValue的最小300倍数
val ff = df.withColumn("value_new", 
  ceil(col("slotValue") / 300.0) * 300
)
display(ff)

逻辑说明：ceil(col("slotValue")/300.0)对数值向上取整得到倍数，再乘以300即可得到目标值，全程分布式处理，无Driver端性能瓶颈。

方案二：改造为UDF（保留原逻辑场景） #

如果必须沿用原有的slot列表匹配逻辑，需将函数改造为UDF，同时避免在UDF中执行Spark SQL（Executor端无法直接访问SparkSession）：

// 预先加载slot列表到本地内存
val slots = List(300,600,900,1200,1500,1800,2100,2400,2700,3000,3300,3600).sorted

// 定义UDF
val getnearestFiveMinSlotUdf = udf((next_slot: Int) => {
  slots.filter(_ >= next_slot).headOption.getOrElse(slots.last)
})

val slotValue = List(100,100,100,4,5)
val df = slotValue.toDF("slotValue")
val ff = df.withColumn("value_new", getnearestFiveMinSlotUdf(col("slotValue")))
display(ff)

逻辑说明：将slot列表提前加载到本地，UDF直接在内存中过滤匹配第一个符合条件的值，兼容分布式处理场景。

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