如何在Databricks的DBFS中读取文件自定义元数据last-modified

问题描述 #

我能通过PySpark读取常规元数据，但无法获取自定义创建的last-modified元数据，当前使用的代码如下：

df = spark.read \
  .format("csv") \
  .load("filepath") \
  .select("*", "_metadata")

display(df)

（原提问附带截图展示了last-modified自定义元数据及元数据变量列表）

解决方案 #

默认的_metadata字段仅包含Spark内置元数据，要读取自定义元数据，需根据存储类型采取对应方法：

方法1：读取文件系统级自定义元数据 #

如果last-modified是存储在文件的用户自定义元数据中（如S3/ADLS的文件属性），可以通过Hadoop文件系统API直接获取：

from pyspark.sql.functions import input_file_name, lit

# 为每条数据关联对应的文件路径
df_with_path = df.withColumn("file_path", input_file_name())

# 获取Hadoop文件系统实例
fs = spark._jvm.org.apache.hadoop.fs.FileSystem.get(spark._jsc.hadoopConfiguration())

# 定义函数读取单个文件的自定义元数据
def get_last_modified(file_path):
    path = spark._jvm.org.apache.hadoop.fs.Path(file_path)
    status = fs.getFileStatus(path)
    return status.getUserDefinedMetadata().get("last-modified")

# 注册UDF并添加元数据列
get_last_modified_udf = spark.udf.register("get_last_modified", get_last_modified)
df_with_metadata = df_with_path.withColumn("last_modified", get_last_modified_udf("file_path"))

display(df_with_metadata)

方法2：通过Delta Lake读取表级自定义元数据 #

如果数据存储为Delta表，且自定义元数据已写入表属性，可以通过以下方式读取：

# 获取Delta表的元数据
delta_table = spark.read.format("delta").load("delta_table_path")
table_metadata = delta_table._jdf.sparkSession().catalog().getTable("database.table_name").properties()

# 提取自定义元数据
last_modified = table_metadata.get("last-modified")

# 将元数据添加到DataFrame
df_with_metadata = df.withColumn("last_modified", lit(last_modified))

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