Spark中注册Iceberg Day分区转换UDF遇到的类型转换问题

报错原因 #

Iceberg的Transforms.day针对带时区的Timestamp类型转换时，实际接收的是时间戳的毫秒数（Long类型），而非直接的java.sql.Timestamp对象。直接传递Timestamp实例会触发类型强制转换错误。

修复后的代码实现 #

方式1：UDF接收Long类型参数 #

直接将UDF的输入参数改为Long（对应Timestamp的毫秒数），复用Iceberg的转换逻辑：

import org.apache.iceberg.transforms.Transforms
import org.apache.iceberg.types.Types

// 初始化带时区的Day转换逻辑
val dayTransform = Transforms.day[Long](Types.TimestampType.withZone())

// 注册UDF：接收时间戳毫秒数，返回自1970-01-01以来的天数
val dayUdf = spark.udf.register("ts_day", (tsMillis: Long) => dayTransform.apply(tsMillis))

// 使用时将Timestamp列转为Long类型传入
df.withColumn("dayPartition", expr(s"ts_day(${timestampCol} cast long)"))

方式2：UDF直接处理Timestamp对象 #

如果希望UDF参数保持为Timestamp，可在内部转为毫秒数后调用Iceberg逻辑：

import org.apache.iceberg.transforms.Transforms
import org.apache.iceberg.types.Types
import java.sql.Timestamp

val dayTransform = Transforms.day[Long](Types.TimestampType.withZone())

// 注册UDF：接收Timestamp对象，内部转毫秒数后执行转换
val dayUdf = spark.udf.register("ts_day", (ts: Timestamp) => {
  dayTransform.apply(ts.getTime)
})

// 直接传入Timestamp列即可使用
df.withColumn("dayPartition", expr(s"ts_day(${timestampCol})"))

Year转换的复用实现 #

Year转换可套用完全相同的逻辑：

import org.apache.iceberg.transforms.Transforms
import org.apache.iceberg.types.Types
import java.sql.Timestamp

val yearTransform = Transforms.year[Long](Types.TimestampType.withZone())

val yearUdf = spark.udf.register("ts_year", (ts: Timestamp) => {
  yearTransform.apply(ts.getTime)
})

df.withColumn("yearPartition", expr(s"ts_year(${timestampCol})"))

逻辑一致性说明 #

以上实现完全复用Iceberg内置的分区转换逻辑，和Iceberg表定义中partition by day(timestamp_col)或partition by year(timestamp_col)的计算结果完全一致，无需自行实现日期截断逻辑。

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