Databricksにおけるファイルメタデータカラム

[File metadata column \| Databricks on AWS](https://docs.databricks.com/ingestion/file-metadata-column.html) [2022/6/9時点]の翻訳です。

:::note warn
本書は抄訳であり内容の正確性を保証するものではありません。正確な内容に関しては原文を参照ください。
:::

:::note info
**注意**
Databricksランタイム10.5以降で使用できます。
:::

`_metadata`カラムを用いることで、入力ファイルのメタデータ情報を取得することができます。`_metadata`カラムは*隠し*カラムであり、すべてのファイルフォマットに対して利用することができます。返却されるデータフレームに`_metadata`カラムを含めるには、クエリーの中で明示的に参照する必要があります。

データソースに`_metadata`というカラムが含まれている場合、クエリーはデータソースからのカラムを返却し、ファイルメタデータを返却しません。

:::note warn
**警告！**
将来のリリースでは`_metadata`カラムに新たなフィールドが追加される場合があります。`_metadata`カラムがアップデートされた際のスキーマエボリューションのエラーを防ぐには、お使いのクエリーのカラムで特定のフィールドを選択するようにすることをお勧めします。[サンプル](#サンプル)をご覧ください。
:::

# サポートされるメタデータ

`_metadata`カラムは以下のフィールドを持つ`STRUCT`です。

| 名前 | タイプ | 説明 | サンプル |
|:--|:--|:--|:--|
| file_path   | STRING  | 入力ファイルのファイルパス。  | `file:/tmp/f0.csv`  |
| file_name  |  STRING | 入力ファイルの名前と拡張子。  | `f0.csv`  |
|  file_size | LONG  | 入力ファイルのバイト長。  | `628`  |
|  file_modification_time |  TIMESTAMP | 入力ファイルの最終更新時刻タイムスタンプ。  | `2021-12-20 20:05:21`  |

# サンプル

## 基本的なファイルベースのデータソースのリーダーでの使用

```py: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_modification_time": "2021-07-02 01:05:21" |
|         |     | }                                                  |
+---------+-----+----------------------------------------------------+
|         |     | {                                                  |
|         |     |    "file_path": "dbfs:/tmp/f1.csv",                |
| Frank   | 24  |    "file_name": "f1.csv",                          |
|         |     |    "file_size": 12,                                |
|         |     |    "file_modification_time": "2021-12-20 02:06:21" |
|         |     | }                                                  |
+---------+-----+----------------------------------------------------+
'''
```

```scala: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_modification_time": "2021-07-02 01:05:21" |
|         |     | }                                                  |
+---------+-----+----------------------------------------------------+
|         |     | {                                                  |
|         |     |    "file_path": "dbfs:/tmp/f1.csv",                |
| Frank   | 24  |    "file_name": "f1.csv",                          |
|         |     |    "file_size": 10,                                |
|         |     |    "file_modification_time": "2021-12-20 02:06:21" |
|         |     | }                                                  |
+---------+-----+----------------------------------------------------+
*/
```

## 特定のフィールドの選択

```py:Python
spark.read \
  .format("csv") \
  .schema(schema) \
  .load("dbfs:/tmp/*") \
  .select("_metadata.file_name", "_metadata.file_size")
```

```scala:Scala
spark.read
  .format("csv")
  .schema(schema)
  .load("dbfs:/tmp/*")
  .select("_metadata.file_name", "_metadata.file_size")
```

## フィルターでの使用

```py:Python
spark.read \
  .format("csv") \
  .schema(schema) \
  .load("dbfs:/tmp/*") \
  .select("*") \
  .filter(col("_metadata.file_name") == lit("test.csv"))
```

```scala:Scala
spark.read
  .format("csv")
  .schema(schema)
  .load("dbfs:/tmp/*")
  .select("*")
  .filter(col("_metadata.file_name") === lit("test.csv"))
```

## COPY INTOでの使用

```sql:SQL
COPY INTO my_delta_table
FROM (
  SELECT *, _metadata FROM 's3://my-bucket/csvData'
)
FILEFORMAT = CSV
```

## Auto Loaderでの使用

```py:Python
spark.readStream \
  .format("cloudFiles") \
  .option("cloudFiles.format", "csv") \
  .schema(schema) \
  .load("s3://my-bucket/csvData") \
  .select("*", "_metadata") \
  .writeStream \
  .format("delta") \
  .option("checkpointLocation", checkpointLocation) \
  .start(targetTable)
```

```scala:Scala
spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "csv")
  .schema(schema)
  .load("s3://my-bucket/csvData")
  .select("*", "_metadata")
  .writeStream
  .format("delta")
  .option("checkpointLocation", checkpointLocation)
  .start(targetTable)
```

# 関連資料

- [COPY INTO](https://docs.databricks.com/spark/latest/spark-sql/language-manual/delta-copy-into.html)
- [DatabricksのAuto Loader](https://qiita.com/taka_yayoi/items/df143647dcf5942b51c6)
- [Structured Streaming](https://docs.databricks.com/spark/latest/structured-streaming/index.html)

### Databricks 無料トライアル

[Databricks 無料トライアル](https://databricks.com/jp/try-databricks)