Scala 2.11下Spark DataFrame用withColumn添加Date列报错解决

这个问题我之前也碰到过，核心原因是Spark的withColumn方法对第二个参数的类型要求很明确——它需要的是一个Column表达式，而不是直接的Java对象，所以你直接传new Date()会触发类型不匹配的错误。下面给你两种可行的解决方案：

方案一：使用Spark内置函数生成当前时间（推荐） #

利用Spark的current_timestamp()函数生成当前时间戳，再转换为DateType（对应Java的java.util.Date类型），这种方式适合需要在任务执行阶段获取当前时间的场景：

// 导入必要的包
import org.apache.spark.sql.functions._
import org.apache.spark.sql.types.DateType

// 读取原表数据（你的原有代码）
val inputDf = sqlContext.read
  .format("jdbc")
  .option("url", "jdbc://blah://whatever")
  .option("query", "SELECT * FROM input WHERE id < 500")
  .load()

// 添加lastRanOn列
val updatedDf = inputDf.withColumn("lastRanOn", current_timestamp().cast(DateType))

方案二：手动创建Date对象并包装为Column #

如果你需要的是Driver端启动时的固定时间（而非任务执行时的时间），可以先创建java.util.Date实例，再用lit()函数把它包装成Column，最后转换类型：

// 导入必要的包
import org.apache.spark.sql.functions.lit
import java.util.Date
import org.apache.spark.sql.types.DateType

// 读取原表数据（你的原有代码）
val inputDf = sqlContext.read
  .format("jdbc")
  .option("url", "jdbc://blah://whatever")
  .option("query", "SELECT * FROM input WHERE id < 500")
  .load()

// 创建当前时间对象，包装为Column并添加到DataFrame
val currentDate = new Date()
val updatedDf = inputDf.withColumn("lastRanOn", lit(currentDate).cast(DateType))

两种方案的区别 #

• 方案一的current_timestamp()会在Spark任务执行时生成时间，适合分布式场景下需要统一执行时间的需求；
• 方案二的时间是在Driver进程启动时生成的，会被广播到所有Executor节点，适合需要固定任务启动时间的场景。

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