定义Flink SQL函数时指定任意数据类型报错，求解决方法

问题原因 #

Flink的ScalarFunction依赖明确的类型信息完成序列化、类型检查与执行计划优化，java.lang.Object作为泛型根类，Flink无法自动推断其具体数据类型，因此抛出该错误。

修复方案 #

方案1：针对具体数据类型实现（推荐） #

如果业务场景仅需处理特定类型数据（如String、Integer等），直接将参数替换为具体类型即可，Flink能自动识别并处理：

class HashCodeFunction2 extends ScalarFunction {
  def eval(s: String): Int = {
    if (s == null) 0 else s.hashCode()
  }

  // 可重载方法支持多种类型
  def eval(i: Integer): Int = {
    if (i == null) 0 else i.hashCode()
  }
}

方案2：使用RAW类型处理任意Object类型 #

若必须支持任意Object类型，需通过@TypeHint注解显式指定RAW类型，告知Flink将参数视为未解析的原始类型：

import org.apache.flink.table.functions.ScalarFunction
import org.apache.flink.api.common.typeinfo.TypeInformation
import org.apache.flink.table.api.dataview.TypeHint

class HashCodeFunction2 extends ScalarFunction {
  @TypeHint("RAW")
  def eval(s: Object): Int = {
    if (s == null) 0 else s.hashCode()
  }

  // 可选：注册函数时手动指定类型信息
  override def getParameterTypes(signature: Array[Class[_]]): Array[TypeInformation[_]] = {
    Array(TypeInformation.of(new TypeHint[Object]() {}))
  }
}

注册函数时也可显式指定参数类型：

tableEnv.createTemporarySystemFunction("hash_code", HashCodeFunction2.getClass)

注意事项 #

• 优先选择方案1，具体类型能让Flink生成更优的执行计划，避免序列化性能损耗。
• 使用RAW类型时，需确保输入对象可序列化，否则会引发序列化失败问题。

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