Kolom met bestandsmetagegevens

U kunt metagegevensgegevens voor invoerbestanden ophalen met de kolom _metadata. De _metadata kolom is een verborgen kolom en is beschikbaar voor alle invoerbestandsindelingen. Als u de _metadata kolom in het geretourneerde DataFrame wilt opnemen, moet u deze expliciet selecteren in de leesquery waarin u de bron opgeeft.

Als de gegevensbron een kolom met de naam _metadatabevat, retourneren query's de kolom uit de gegevensbron en niet de metagegevens van het bestand.

Waarschuwing

Nieuwe velden kunnen in toekomstige releases aan de _metadata kolom worden toegevoegd. Om fouten in de ontwikkeling van schema's te voorkomen als de _metadata kolom wordt bijgewerkt, raadt Databricks aan om specifieke velden te selecteren in de kolom in uw query's. Zie voorbeelden.

Ondersteunde metagegevens

De _metadata kolom is een STRUCT en bevat de volgende velden:

Naam Typologie Beschrijving Voorbeeld Minimale release van Databricks Runtime
file_path STRING Bestandspad van het invoerbestand. file:/tmp/f0.csv 10.5
file_name STRING De naam van het invoerbestand en de extensie. f0.csv 10.5
file_size LONG De lengte van het invoerbestand, in bytes. 628 10.5
bestand_wijzigingstijd TIMESTAMP Laatste wijzigingstijdstempel van het invoerbestand. 2021-12-20 20:05:21 10.5
file_block_start LONG Beginverschil van het blok dat wordt gelezen, in bytes. 0 13,0
bestand_blok_lengte LONG De lengte van het blok dat wordt gelezen, in bytes. 628 13,0

Zie de kolom Objectmetagegevens voor het ophalen van aanvullende eigenschappen op cloudobjectniveau.

Voorbeelden

Gebruiken in een eenvoudige gegevensbronlezer op basis van bestanden

Python

df = spark.read \
  .format("csv") \
  .schema(schema) \
  .load("/Volumes/catalog_name/schema_name/volume_name/data/*") \
  .select("*", "_metadata")

display(df)

'''
Result:
+---------+-----+----------------------------------------------------+
|   name  | age |                 _metadata                          |
+=========+=====+====================================================+
|         |     | {                                                  |
|         |     |    "file_path": "/Volumes/catalog_name/            |
| Debbie  | 18  |      schema_name/volume_name/data/f0.csv",         |
|         |     |    "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": "/Volumes/catalog_name/            |
| Frank   | 24  |      schema_name/volume_name/data/f1.csv",         |
|         |     |    "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("/Volumes/catalog_name/schema_name/volume_name/data/*")
  .select("*", "_metadata")

display(df)

/* Result:
+---------+-----+----------------------------------------------------+
|   name  | age |                 _metadata                          |
+=========+=====+====================================================+
|         |     | {                                                  |
|         |     |    "file_path": "/Volumes/catalog_name/            |
| Debbie  | 18  |      schema_name/volume_name/data/f0.csv",         |
|         |     |    "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": "/Volumes/catalog_name/            |
| Frank   | 24  |      schema_name/volume_name/data/f1.csv",         |
|         |     |    "file_name": "f1.csv",                          |
|         |     |    "file_size": 10,                                |
|         |     |    "file_block_start": 0,                          |
|         |     |    "file_block_length": 12,                        |
|         |     |    "file_modification_time": "2021-12-20 02:06:21" |
|         |     | }                                                  |
+---------+-----+----------------------------------------------------+
*/

Specifieke velden selecteren

Python

spark.read \
  .format("csv") \
  .schema(schema) \
  .load("/Volumes/catalog_name/schema_name/volume_name/data/*") \
  .select("_metadata.file_name", "_metadata.file_size")

Scala

spark.read
  .format("csv")
  .schema(schema)
  .load("/Volumes/catalog_name/schema_name/volume_name/data/*")
  .select("_metadata.file_name", "_metadata.file_size")

Gebruiken in filters

Python

spark.read \
  .format("csv") \
  .schema(schema) \
  .load("/Volumes/catalog_name/schema_name/volume_name/data/*") \
  .select("*") \
  .filter(col("_metadata.file_name") == lit("test.csv"))

Scala

spark.read
  .format("csv")
  .schema(schema)
  .load("/Volumes/catalog_name/schema_name/volume_name/data/*")
  .select("*")
  .filter(col("_metadata.file_name") === lit("test.csv"))

Gebruiken in COPY INTO (verouderd)

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

Gebruiken in automatisch laadprogramma

Als uw brongegevens een kolom met de naam _metadatabevatten, wijzigt u de naam ervan in source_metadata. Als u de naam ervan niet wijzigt, hebt u geen toegang tot de kolom met bestandsmetagegevens in de doeltabel; Query's retourneren in plaats daarvan de bronkolom.

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)

Als u foreachBatch gebruikt en de kolom met bestandsmetagegevens in het streaming DataFrame wilt opnemen, moet u er vóór de foreachBatch functie naar verwijzen in het streaming-read DataFrame. Als u alleen naar de kolom met bestandsmetagegevens in de foreachBatch functie verwijst, wordt de kolom niet opgenomen.

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(...)

Aanvullende informatiebronnen

  • Last updated on