檔案元數據行
您可以使用 資料列取得輸入檔的 _metadata
元數據資訊。 數據_metadata
行是隱藏的數據行,而且適用於所有輸入檔案格式。 若要在傳回的 DataFrame 中包含數據 _metadata
行,您必須在查詢中明確參考它。
如果數據源包含名為 _metadata
的數據行,查詢會從數據源傳回數據行,而不是檔案元數據。
警告
未來版本中可能會將新的欄位新增至 _metadata
數據行。 若要防止數據行更新時 _metadata
發生架構演進錯誤,Databricks 建議從查詢中的數據行選取特定字段。 請參閱 範例。
支援的中繼資料
資料 _metadata
列是 STRUCT
包含下列欄位的 :
名稱 | 類型 | 描述 | 範例 | 最低 Databricks 執行時間版本 |
---|---|---|---|---|
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.windows.net/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.windows.net/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.windows.net/csvData")
.selectExpr("*", "_metadata as source_metadata")
.writeStream
.option("checkpointLocation", checkpointLocation)
.start(targetTable)