Spark调用year、month函数处理Cassandra时间戳为何出现long溢出

• 两者时间戳存储单位不匹配是核心原因：Cassandra的timestamp类型存储的是自1970-01-01 00:00:00 UTC的毫秒数，而Spark内部处理TimestampType时默认会将读取到的毫秒数乘以1000转为微秒存储。一旦Cassandra中存储的时间戳数值超过9223372036854775（即Long.MaxValue / 1000），乘以1000的操作就会触发Long类型溢出，和你遇到的报错栈完全对应。
• 空值过滤不生效是因为溢出发生在Spark读取字段做类型转换的阶段，过滤逻辑还未执行就已经报错。

方案1：读取时指定字段为Long类型手动处理（兼容性最好） #

规避Spark自动做毫秒转微秒的操作，自行控制转换逻辑，同时可以过滤非法时间戳：

import org.apache.spark.sql.cassandra._
import org.apache.spark.sql.types._
import org.apache.spark.sql.functions._

// 自定义schema，仅将logtimestamp指定为LongType，其余字段保持原有类型即可
val customSchema = StructType(Seq(
  StructField("logtimestamp", LongType, nullable = true),
  // 此处补充表其余字段的schema定义，示例：StructField("content", StringType, nullable = true)
))

val logsDF = spark.read
  .cassandraFormat("tableName", "cw")
  .schema(customSchema)
  .load()

val newlogs = logsDF
  // 可根据业务实际时间范围过滤非法脏数据，示例范围为1970年~2100年
  .filter(col("logtimestamp").between(-62135798400000L, 4102444800000L))
  // 手动将毫秒时间戳转为Spark的Timestamp类型
  .withColumn("logtimestamp_ts", to_timestamp(col("logtimestamp") / 1000))
  .withColumn("year", year(col("logtimestamp_ts")))
  .withColumn("month", month(col("logtimestamp_ts")))
  // 可选：删除不需要的中间转换字段
  .drop("logtimestamp_ts")

newlogs.write.cassandraFormat("tableName_v2", "cw")
 .mode("Append").save()

方案2：修改Spark Connector配置（适合Datastax Connector 3.0+版本） #

启动spark-shell时添加如下参数，指定Cassandra时间戳以毫秒为单位读取，无需自动转微秒：

--conf spark.cassandra.sql.timestampType=millis

配置后原有代码无需修改即可正常运行。

前置验证方法 #

可先读取所有logtimestamp的极值确认是否存在非法值：

logsDF.select(max("logtimestamp"), min("logtimestamp")).show()

如果查询到的最大值大于9223372036854775，即可确认是脏数据导致的溢出问题。

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