Columna de metadatos de archivo

Puede obtener información de metadatos para los archivos de entrada con la columna _metadata. La columna _metadata es una columna oculta y está disponible para todos los formatos de archivo de entrada. Para incluir la _metadata columna en el dataframe devuelto, debe seleccionarla explícitamente en la consulta de lectura donde especifique el origen.

Si el origen de datos contiene una columna denominada _metadata, las consultas devuelven la columna desde el origen de datos y no los metadatos del archivo.

Advertencia

Se pueden agregar nuevos campos a la columna _metadata en futuras versiones. Para evitar errores de evolución del esquema si se actualiza la columna _metadata, Databricks recomienda seleccionar campos específicos de la columna en las consultas. Consulte ejemplos.

Metadatos admitidos

La columna _metadata es un objeto STRUCT que contiene los campos siguientes:

Nombre Tipo Descripción Ejemplo Versión mínima de Databricks Runtime
file_path STRING Ruta de acceso del archivo de entrada. file:/tmp/f0.csv 10.5
file_name STRING Nombre del archivo de entrada junto con su extensión. f0.csv 10.5
file_size LONG Tamaño del archivo de entrada, en bytes. 628 10.5
hora_de_modificacion_del_archivo TIMESTAMP Marca de tiempo de la última modificación del archivo de entrada. 2021-12-20 20:05:21 10.5
inicio_del_bloque_de_archivo LONG Desplazamiento inicial del bloque que se va a leer, en bytes. 0 13,0
longitud_del_bloque_de_archivo LONG Longitud del bloque que se lee, en bytes. 628 13,0

Para obtener propiedades adicionales de nivel de objeto en la nube, consulte Columna metadatos de objeto.

Ejemplos

Uso en un lector de origen de datos basado en archivos básico

Pitón

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

Selección de campos específicos

Pitón

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

Uso en filtros

Pitón

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

Uso en COPY INTO (heredado)

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

Uso en el cargador automático

Si los datos de origen contienen una columna denominada _metadata, cámbiela por source_metadata. Si no cambia el nombre, no puede acceder a la columna de metadatos del archivo en la tabla de destino; las consultas devolverán la columna de origen en su lugar.

Pitón

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)

Si usa foreachBatch y desea incluir la columna de metadatos de archivo en el DataFrame de streaming, debe hacer referencia a esta columna en el DataFrame de lectura de streaming antes de la función foreachBatch. Si solo hace referencia a la columna de metadatos de archivo dentro de la foreachBatch función , la columna no se incluye.

Pitón

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

Recursos adicionales

  • Last updated on