Поделиться через

Столбец метаданных файла

Вы можете получить сведения о метаданных для входных файлов с помощью столбца _metadata . Столбец _metadata является скрытым столбцом и доступен для всех форматов входных файлов. Чтобы включить _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
время_изменения_файла TIMESTAMP Временная метка последнего изменения входного файла. 2021-12-20 20:05:21 10,5
file_block_start LONG Начальное смещение блока, считываемого в байтах. 0 13,0
длина_блока_файла LONG Длина блока, который читается, в байтах. 628 13,0

Примеры

Использование в базовом средстве чтения источников данных на основе файлов

Питон

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

Выбор определенных полей

Питон

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

Использование в фильтрах

Питон

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://[email protected]/csvData'
)
FILEFORMAT = CSV

Использование в автозагрузчике

Если исходные данные содержат столбец с именем _metadata, переименуйте его source_metadataв . Если вы не переименовываете его, вы не можете получить доступ к столбцу метаданных файла в целевой таблице; запросы возвращают исходный столбец.

Питон

spark.readStream \
  .format("cloudFiles") \
  .option("cloudFiles.format", "csv") \
  .schema(schema) \
  .load("abfss://[email protected]/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://[email protected]/csvData")
  .selectExpr("*", "_metadata as source_metadata")
  .writeStream
  .option("checkpointLocation", checkpointLocation)
  .start(targetTable)

Если вы используете foreachBatch и хотите включить столбец метаданных файла в потоковый DataFrame, необходимо ссылаться на него в DataFrame потокового чтения перед функцией foreachBatch. Если вы указываете только на столбец метаданных файла внутри функции foreachBatch, столбец не будет включен.

Питон

spark.readStream \
  .format("cloudFiles") \
  .option("cloudFiles.format", "csv") \
  .load("abfss://[email protected]/csvData") \
  .select("*", "metadata") \
  .writeStream \
  .foreachBatch(...)

язык программирования Scala

spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "csv")
  .load("abfss://[email protected]/csvData")
  .select("*", "metadata")
  .writeStream
  .foreachBatch(...)