Spark读取HDFS中Parquet文件时，如何为每条记录添加文件创建时间列？

你之前用current_timestamp()得到的是执行该操作的当前时间，而非源文件的创建时间，所以不符合预期。下面是正确的实现方案：

方法一：通过HDFS文件元数据获取（适合已存在的文件） #

步骤1：读取文件并添加源文件路径列 #

首先用input_file_name()函数为每条记录关联对应的源文件路径：

import org.apache.spark.sql.functions.input_file_name

val df = spark.read.parquet("/data/module/")
  .withColumn("source_file", input_file_name())

步骤2：创建UDF获取HDFS文件的创建时间 #

HDFS文件的modification time（修改时间）在文件写入后未被修改的情况下，等同于创建时间。我们可以通过Hadoop API获取该时间并转换为Timestamp：

import org.apache.spark.sql.functions.udf
import org.apache.hadoop.fs.{FileSystem, Path}
import java.sql.Timestamp

val getFileCreationTime = udf((filePath: String) => {
  val fs = FileSystem.get(spark.sparkContext.hadoopConfiguration)
  val path = new Path(filePath)
  if (fs.exists(path)) {
    new Timestamp(fs.getFileStatus(path).getModificationTime)
  } else {
    null
  }
})

步骤3：添加文件创建时间列 #

调用UDF生成目标列：

val dfWithCreationTime = df.withColumn("records_inserted_time", getFileCreationTime($"source_file"))

方法二：写入时提前嵌入时间（性能更优） #

如果是你自己写入的Parquet文件，建议在写入时直接把创建时间作为数据列写入，后续读取时无需额外操作：

// 写入时添加时间列
val dfToWrite = originalDF.withColumn("records_inserted_time", current_timestamp())
dfToWrite.write.parquet("/data/module/file1.parquet")

// 读取时直接使用该列
val df = spark.read.parquet("/data/module/")

注意事项 #

• 方法一中，若文件后续被修改过，modification time会更新，此时获取的不是原始创建时间；若需要严格的原始创建时间，优先使用方法二。
• 确保Spark集群的Hadoop配置正确，能够正常访问目标HDFS路径。

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