Sdílet prostřednictvím

Sloupec metadat souboru

Informace o metadatech pro vstupní soubory můžete získat pomocí sloupce _metadata. Sloupec _metadata je skrytý sloupec a je k dispozici pro všechny formáty vstupních souborů. Pokud chcete sloupec zahrnout _metadata do vráceného datového rámce, musíte ho explicitně vybrat v dotazu pro čtení, kde zadáte zdroj.

Pokud zdroj dat obsahuje sloupec s názvem _metadata, dotazy vrátí sloupec ze zdroje dat, a ne metadata souboru.

Výstraha

Do _metadata sloupce v budoucích verzích se můžou přidat nová pole. Aby se zabránilo chybám vývoje schématu v případě aktualizace sloupce _metadata, doporučuje Databricks vybrat konkrétní pole ze sloupce v dotazech. Podívejte se na příklady.

Podporovaná metadata

Sloupec _metadata je STRUCT obsahující tyto pole:

Název Typ Popis Příklad Minimální verze Databricks Runtime
file_path STRING Cesta k vstupnímu souboru. file:/tmp/f0.csv 10.5
file_name STRING Název vstupního souboru spolu s jeho příponou. f0.csv 10.5
file_size LONG Délka vstupního souboru v bajtech 628 10.5
čas_upravy_souboru TIMESTAMP Časové razítko poslední změny vstupního souboru 2021-12-20 20:05:21 10.5
file_block_start LONG Počáteční posun bloku, který se čte, v bajtech 0 13.0
délka_bloku_souboru LONG Délka čteného bloku v bajtech. 628 13.0

Příklady

Použití v základní čtečce zdrojů dat založených na souborech

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" |
|         |     | }                                                  |
+---------+-----+----------------------------------------------------+
*/

Výběr konkrétních polí

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")

Použití ve filtrech

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"))

Použití v COPY INTO (původní systém)

COPY INTO my_delta_table
FROM (
  SELECT *, _metadata FROM 'abfss://my-container-name@storage-account-name.dfs.core.windows.net/csvData'
)
FILEFORMAT = CSV

Použití v automatickém zavaděči

Pokud zdrojová data obsahují sloupec s názvem _metadata, přejmenujte je na source_metadata. Pokud ho nepřejmenujete, nemůžete získat přístup ke sloupci metadat souborů v cílové tabulce; dotazy místo toho vrátí zdrojový sloupec.

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)

Pokud používáte foreachBatch a chcete zahrnout sloupec metadat souboru do streamovaného datového rámce, musíte jej uvést v datovém rámci pro čtení streamovaných dat před funkcí foreachBatch. Pokud odkazujete pouze na sloupec metadat souboru uvnitř foreachBatch funkce, sloupec se nezahrne.

Python

spark.readStream \
  .format("cloudFiles") \
  .option("cloudFiles.format", "csv") \
  .load("abfss://my-container-name@storage-account-name.dfs.core.windows.net/csvData") \
  .select("*", "metadata") \
  .writeStream \
  .foreachBatch(...)

Scala

spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "csv")
  .load("abfss://my-container-name@storage-account-name.dfs.core.windows.net/csvData")
  .select("*", "metadata")
  .writeStream
  .foreachBatch(...)
  • Last updated on