文件元数据列
可以使用 _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-container-name@storage-account-name.dfs.core.chinacloudapi.cn/csvData'
)
FILEFORMAT = CSV
在自动加载程序中使用
注意
编写 _metadata
列时,我们将其重命名为 source_metadata
。 将其编写为 _metadata
将无法访问目标表中的元数据列,因为如果数据源包含名为 _metadata
的列,查询将从数据源返回该列,而不是文件元数据。
Python
spark.readStream \
.format("cloudFiles") \
.option("cloudFiles.format", "csv") \
.schema(schema) \
.load("abfss://my-container-name@storage-account-name.dfs.core.chinacloudapi.cn/csvData") \
.selectExpr("*", "_metadata as source_metadata") \
.writeStream \
.option("checkpointLocation", checkpointLocation) \
.start(targetTable)
Scala
spark.readStream
.format("cloudFiles")
.option("cloudFiles.format", "csv")
.schema(schema)
.load("abfss://my-container-name@storage-account-name.dfs.core.chinacloudapi.cn/csvData")
.selectExpr("*", "_metadata as source_metadata")
.writeStream
.option("checkpointLocation", checkpointLocation)
.start(targetTable)