Spark新增Timestamp列触发Column不可序列化错误问题咨询

问题根因 #

你编写的新增time_stamp列的代码本身没有语法错误，报错核心原因是构造source DataFrame时，直接将org.apache.spark.sql.Column类型的对象放入了Row中。
Spark的Row是存储业务数据的容器，仅支持存入可序列化的普通JVM类型（Int、String、Long、Timestamp等）。Column是Spark SQL的算子抽象类，未实现序列化接口，仅能在驱动端构造执行计划时使用，不能作为数据值存入Row。当作业提交后需要将分区数据序列化发送到Executor执行时，就会抛出NotSerializableException异常。

解决方案 #

方案1：修正Source DataFrame的构造逻辑 #

所有存入Row的内容都替换为普通可序列化值，如果有生成时间戳的需求，优先通过withColumn统一新增，不要提前把current_timestamp()这类Column放入Row：

// 错误写法：直接将Column放入Row
val rawRows = Seq(Row(1, "业务数据1", current_timestamp()))
// 正确写法1：构造Row时直接生成实际时间值
val rawRows = Seq(Row(1, "业务数据1", System.currentTimeMillis()))
// 正确写法2：Row不预存时间，构造完DataFrame后统一新增时间列
val rawRows = Seq(Row(1, "业务数据1"))
val source = spark.createDataFrame(rawRows, yourSchema)
val modified = source.withColumn("time_stamp", current_timestamp())
modified.show()

方案2：表达式存储改用字符串 #

如果确实需要在数据中存储表达式逻辑，不要用Column类型，改为存储字符串格式的表达式内容，后续需要执行时再通过expr()函数转换为Column即可：

// 存表达式的时候用字符串
val rawRows = Seq(Row(1, "业务数据1", "current_timestamp()"))
// 后续需要用的时候再转成Column
val modified = source.withColumn("calculated_value", expr(col("expr_str").toString))

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