Colonne de métadonnées de fichier

Vous pouvez obtenir des informations de métadonnées pour les fichiers d’entrée avec la colonne _metadata. La colonne _metadata est une colonne masquée et est disponible pour tous les formats de fichier d’entrée. Pour inclure la colonne _metadata dans le DataFrame retourné, vous devez la référencer explicitement dans votre requête.

Si la source de données contient une colonne nommée _metadata, les requêtes retournent la colonne depuis la source de données, plutôt que les métadonnées du fichier.

Avertissement

De nouveaux champs peuvent être ajoutés à la colonne _metadata dans les versions futures. Pour éviter des erreurs d’évolution du schéma si la colonne _metadata est mise à jour, Databricks recommande de sélectionner des champs spécifiques de la colonne dans vos requêtes. Consultez les exemples.

Métadonnées prises en charge

La colonne _metadata est un STRUCT contenant les champs suivants :

Nom Type Description Exemple Version minimale de Databricks Runtime
file_path STRING Chemin d’accès du fichier d’entrée. file:/tmp/f0.csv 10.5
file_name STRING Nom du fichier d’entrée avec son extension. f0.csv 10.5
file_size LONG Longueur du fichier d’entrée, en octets. 628 10.5
file_modification_time TIMESTAMP Horodatage de la dernière modification du fichier d’entrée. 2021-12-20 20:05:21 10.5
file_block_start LONG Décalage de début du bloc en cours de lecture, en octets. 0 13.0
file_block_length LONG Longueur du bloc en cours de lecture, en octets. 628 13.0

Exemples

Utiliser dans un lecteur de source de données basé sur un fichier de base

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

Sélectionner des champs spécifiques

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

Utiliser dans des filtres

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

Utiliser dans 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

Utiliser dans Chargeur automatique

Notes

Lors de l’écriture de la colonne _metadata, nous la renommons source_metadata. L’écrire en tant que _metadata rend impossible l’accès à la colonne de métadonnées dans la table cible, car si la source de données contient une colonne nommée _metadata, les requêtes retournent la colonne de la source de données et pas les métadonnées du fichier.

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)