Spark Streaming如何获取源文件修改时间以计算写入时延？

1. Spark是否提供内置函数获取源文件修改时间？ #

Spark没有直接返回对象存储文件修改时间的内置SQL函数，但可以通过启用文件数据源元数据列的方式高效获取，完全避免重复调用对象存储接口的成本。

只需在读取数据前设置配置spark.sql.files.metadataColumnEnabled为true，就能直接使用两个内置元数据列：

• _metadata_file_path：等价于input_file_name()的返回值
• _metadata_file_modification_time：文件的毫秒级修改时间戳

这个元数据是Spark扫描文件列表时就已获取的，不需要额外发起Head请求，完美解决单文件百万记录重复请求的问题。示例代码：

// Scala 示例
spark.conf.set("spark.sql.files.metadataColumnEnabled", "true")

val streamDf = spark.readStream
  .format("parquet") // 替换为你的文件格式（CSV/JSON等）
  .load("s3://your-bucket/path/")
  .select(
    "*",
    "_metadata_file_path",
    "_metadata_file_modification_time"
  )

2. 能否从检查点的timestamp获取文件时间？ #

你在_checkpoints/sources/0中看到的timestamp字段，确实是Spark发现新文件时记录的文件最后修改时间，但Spark并没有提供官方API让业务逻辑直接读取检查点中的这个值——检查点的核心作用是故障恢复，并非业务数据来源。

但你完全不需要依赖检查点：上述元数据列方案已经能在数据流处理过程中直接获取该时间戳，比读取检查点更可靠、更高效。

计算时间差的实现 #

拿到文件修改时间后，结合写入前的当前时间计算差值的示例：

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

val finalDf = streamDf
  .withColumn("write_time", current_timestamp())
  .withColumn(
    "time_diff_seconds",
    unix_timestamp(col("write_time")) - (col("_metadata_file_modification_time") / 1000)
  )

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