文件元数据列

  • 项目

可以使用 _metadata 列获取输入文件的元数据信息。 _metadata 列是隐藏列，可用于所有输入文件格式。 若要在返回的数据帧中包含 _metadata 列，必须在查询中显式引用该列。

如果数据源包含名为 _metadata 的列，则查询将从数据源返回该列，而不是文件元数据。

警告

在将来的版本中，新字段可能会添加到 _metadata 列中。 为了防止在更新 _metadata 列时出现架构演变错误，Databricks 建议从查询中的列中选择特定字段。 请参阅示例。

支持的元数据

_metadata 列是包含以下字段的 STRUCT：

名称 Type 说明 示例 最低 Databricks Runtime 版本
file_path STRING 输出文件的文件路径。 file:/tmp/f0.csv 10.5
file_name STRING 输入文件的名称及其扩展名。 f0.csv 10.5
file_size LONG 输入文件的长度（以字节为单位）。 628 10.5
file_modification_time TIMESTAMP 输入文件的上次修改时间戳。 2021-12-20 20:05:21 10.5
file_block_start LONG 读取的块的起始偏移量（以字节为单位）。 0 13.0
file_block_length LONG 读取的块的长度（以字节为单位）。 628 13.0

示例

在基于文件的基本数据源读取器中使用

Python

df = spark.read \
  .format("csv") \
  .schema(schema) \
  .load("dbfs:/tmp/*") \
  .select("*", "_metadata")

display(df)

'''
Result:
+---------+-----+----------------------------------------------------+
|   name  | age |                 _metadata                          |
+=========+=====+====================================================+
|         |     | {                                                  |
|         |     |    "file_path": "dbfs:/tmp/f0.csv",                |
| Debbie  | 18  |    "file_name": "f0.csv",                          |
|         |     |    "file_size": 12,                                |
|         |     |    "file_block_start": 0,                          |
|         |     |    "file_block_length": 12,                        |
|         |     |    "file_modification_time": "2021-07-02 01:05:21" |
|         |     | }                                                  |
+---------+-----+----------------------------------------------------+
|         |     | {                                                  |
|         |     |    "file_path": "dbfs:/tmp/f1.csv",                |
| Frank   | 24  |    "file_name": "f1.csv",                          |
|         |     |    "file_size": 12,                                |
|         |     |    "file_block_start": 0,                          |
|         |     |    "file_block_length": 12,                        |
|         |     |    "file_modification_time": "2021-12-20 02:06:21" |
|         |     | }                                                  |
+---------+-----+----------------------------------------------------+
'''

Scala

val df = spark.read
  .format("csv")
  .schema(schema)
  .load("dbfs:/tmp/*")
  .select("*", "_metadata")

display(df_population)

/* Result:
+---------+-----+----------------------------------------------------+
|   name  | age |                 _metadata                          |
+=========+=====+====================================================+
|         |     | {                                                  |
|         |     |    "file_path": "dbfs:/tmp/f0.csv",                |
| Debbie  | 18  |    "file_name": "f0.csv",                          |
|         |     |    "file_size": 12,                                |
|         |     |    "file_block_start": 0,                          |
|         |     |    "file_block_length": 12,                        |
|         |     |    "file_modification_time": "2021-07-02 01:05:21" |
|         |     | }                                                  |
+---------+-----+----------------------------------------------------+
|         |     | {                                                  |
|         |     |    "file_path": "dbfs:/tmp/f1.csv",                |
| Frank   | 24  |    "file_name": "f1.csv",                          |
|         |     |    "file_size": 10,                                |
|         |     |    "file_block_start": 0,                          |
|         |     |    "file_block_length": 12,                        |
|         |     |    "file_modification_time": "2021-12-20 02:06:21" |
|         |     | }                                                  |
+---------+-----+----------------------------------------------------+
*/

选择特定的字段

Python

spark.read \
  .format("csv") \
  .schema(schema) \
  .load("dbfs:/tmp/*") \
  .select("_metadata.file_name", "_metadata.file_size")

Scala

spark.read
  .format("csv")
  .schema(schema)
  .load("dbfs:/tmp/*")
  .select("_metadata.file_name", "_metadata.file_size")

在筛选器中使用

Python

spark.read \
  .format("csv") \
  .schema(schema) \
  .load("dbfs:/tmp/*") \
  .select("*") \
  .filter(col("_metadata.file_name") == lit("test.csv"))

Scala

spark.read
  .format("csv")
  .schema(schema)
  .load("dbfs:/tmp/*")
  .select("*")
  .filter(col("_metadata.file_name") === lit("test.csv"))

在 COPY INTO 中使用

COPY INTO my_delta_table
FROM (
  SELECT *, _metadata FROM 'abfss://my-bucket/csvData'
)
FILEFORMAT = CSV

在自动加载程序中使用

注意

编写 _metadata 列时，我们将其重命名为 source_metadata。 将其编写为 _metadata 将无法访问目标表中的元数据列，因为如果数据源包含名为 _metadata 的列，查询将从数据源返回该列，而不是文件元数据。

Python

spark.readStream \
  .format("cloudFiles") \
  .option("cloudFiles.format", "csv") \
  .schema(schema) \
  .load("abfss://my-bucket/csvData") \
  .selectExpr("*", "_metadata as source_metadata") \
  .writeStream \
  .format("delta") \
  .option("checkpointLocation", checkpointLocation) \
  .start(targetTable)

Scala

spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "csv")
  .schema(schema)
  .load("abfss://my-bucket/csvData")
  .selectExpr("*", "_metadata as source_metadata")
  .writeStream
  .format("delta")
  .option("checkpointLocation", checkpointLocation)
  .start(targetTable)