嗨，我刚好之前踩过这个坑！Spark Cassandra Connector确实不支持你说的"column_a".writeTime这种写法，但它其实提供了专门的内置函数来获取列的写入时间，完全能满足你同时提取原列和对应writeTime的需求，我给你详细讲讲怎么操作：

Spark Cassandra Connector在org.apache.spark.sql.cassandra包下提供了writeTime函数，专门用来获取Cassandra列的写入时间戳。你可以在DataFrame的select操作中，同时指定原列和对应的writeTime计算结果，还能给生成的时间列自定义别名。

具体代码示例 #

Scala版本 #

首先导入必要的依赖函数：

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

然后读取Cassandra表并提取列和对应的写入时间：

// 读取Cassandra表
val cassandraDF = spark.read
  .format("org.apache.spark.sql.cassandra")
  .options(Map(
    "table" -> "your_target_table", 
    "keyspace" -> "your_keyspace"
  ))
  .load()

// 同时选择原列和对应的writeTime
val resultDF = cassandraDF.select(
  col("column_a"),
  writeTime(col("column_a")).alias("column_a_write_time"), // 给时间列起别名
  col("column_b"),
  writeTime(col("column_b")).alias("column_b_write_time"),
  // 可以继续添加更多列和对应的writeTime
  col("column_c"),
  writeTime(col("column_c")).alias("column_c_write_time")
)

resultDF.show()

Python版本 #

Python的导入和写法略有不同，注意要从cassandra的functions模块导入writeTime：

from pyspark.sql import functions as F
from pyspark.sql.cassandra import functions as cassandra_func

# 读取Cassandra表
cassandra_df = spark.read \
    .format("org.apache.spark.sql.cassandra") \
    .options(
        table="your_target_table",
        keyspace="your_keyspace"
    ) \
    .load()

# 提取列和对应的写入时间
result_df = cassandra_df.select(
    F.col("column_a"),
    cassandra_func.writeTime(F.col("column_a")).alias("column_a_write_time"),
    F.col("column_b"),
    cassandra_func.writeTime(F.col("column_b")).alias("column_b_write_time")
)

result_df.show()

额外技巧：转换时间戳为可读格式 #

writeTime函数返回的是毫秒级的Unix时间戳，如果需要转换成人类可读的时间格式，可以配合from_unixtime函数处理（注意要先把毫秒转成秒）：

// Scala示例：添加可读时间列
val resultDFWithReadableTime = cassandraDF.select(
  col("column_a"),
  writeTime(col("column_a")).alias("column_a_write_time"),
  from_unixtime(col("column_a_write_time") / 1000).alias("column_a_write_time_readable")
)

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