Scala环境下Spark Dataframe filter查询使用自定义函数的问题

你定义的hash_id是普通Scala函数，而非Spark可识别的自定义函数（UDF）。当你直接在filter中传入hash_id("ID")时，传入的是"ID"字符串字面量而非DataFrame的ID列对象，函数会直接对字符串"ID"做计算返回Int类型的普通数值，而非Spark的Column类型，因此无法使用Spark SQL专属的===运算符，触发报错。

1. 导入依赖并注册UDF #

首先将普通Scala函数包装为Spark UDF，使其可以作用于DataFrame的列数据：

// 导入udf工具类
import org.apache.spark.sql.functions.udf
// 可选导入，后续可以用$"列名"快速引用列
import spark.implicits._

// 包装自定义逻辑为UDF
val hash_id_udf = udf((id: String) => {
  val two_char = id.takeRight(2).toInt
  val hash_result = two_char % 4
  hash_result
})

2. 在filter中调用UDF #

调用时传入ID列对象，而非字符串字面量：

val filteredDF = DF.filter(hash_id_udf($"ID") === 3)

如果需要在字符串格式的SQL条件中使用该函数，可以将UDF注册到Spark会话中：

spark.udf.register("hash_id", (id: String) => {
  val two_char = id.takeRight(2).toInt
  two_char % 4
})

// 直接用SQL表达式过滤
val filteredDF = DF.where("hash_id(ID) = 3")

需要保证ID列的类型为String，且所有取值的最后两位都是数字，否则toInt方法会抛出类型转换异常，存在脏数据的场景建议提前加校验逻辑处理异常值。

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