共用方式為

檔案中繼資料行

  • 發行項

您可以使用 資料行取得輸入檔的 _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 \
  .format("delta") \
  .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
  .format("delta")
  .option("checkpointLocation", checkpointLocation)
  .start(targetTable)