Databricksワークフローのファイル到着トリガーがGAになりました

こちらの機能がGAになりました！

[**File arrival triggers in Databricks Workflows is GA**](https://docs.databricks.com/en/release-notes/product/2024/february.html#file-arrival-triggers-in-databricks-workflows-is-ga)

> ファイル到着トリガーがすべてのクラウドプロバイダーでGAになりました。このリリースで、既存のUnity Catalog外部ロケーションのサポートに加えて、Unity Catalogのボリュームに新規ファイルが到着した際にDatabricksジョブを実行するためにファイル到着トリガーを活用できるようになります。[新しいファイルが到着したときにジョブをトリガーする](https://docs.databricks.com/ja/workflows/jobs/file-arrival-triggers.html)をご覧ください。

早速試します。

# ボリュームの作成

`/Volumes/takaakiyayoi_catalog/default/landing_zone/`というボリュームを作成します。
![Screenshot 2024-02-14 at 19.39.25.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/7cbedc93-dec9-adb6-db38-3d12d1438aac.png)

# ノートブックの作成

このノートブックは、あとでジョブから呼び出します。[Auto Loader](https://docs.databricks.com/ja/ingestion/auto-loader/index.html)を用いて、新規到着ファイルのみを処理するようにします。`checkpoint_path`や`load`の引数のパスも適宜ボリュームを指定しています。テーブル`takaakiyayoi_catalog.default.ingested`に書き込んでいます。

```py:data_ingest
checkpoint_path = "/Volumes/takaakiyayoi_catalog/default/checkpoint"

(spark.readStream
  .format("cloudFiles") # Auto Loader
  .option("cloudFiles.format", "csv") # ファイルフォーマット
  .option("header", "true")
  .option("delimiter", ",")
  .option("cloudFiles.schemaLocation", checkpoint_path) # スキーマ格納場所
  .load("/Volumes/takaakiyayoi_catalog/default/landing_zone") # 読み込みボリュームパス
  .writeStream
  .option("checkpointLocation", checkpoint_path) # チェックポイント格納場所
  .trigger(availableNow=True) # 増分バッチ処理
  .toTable("takaakiyayoi_catalog.default.ingested")) # 書き込み先テーブル
```

# ジョブの作成

タスクを作成し、上のノートブックを指定します。トリガータイプを**ファイル到着**にし、上記ボリュームのパスを指定します。
![Screenshot 2024-02-14 at 19.40.39.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/af6e6d5e-36f9-5208-51f0-b4858b435e03.png)
![Screenshot 2024-02-14 at 19.41.17.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/0a8c0ad4-8efa-9eec-8ec6-0032a8c6b6d5.png)


これでファイル到着によって起動するジョブの準備ができました。

# ファイルのアップロード

以下のようなCSVファイルを準備します。

```001.csv
id,first,last
1, taka, yayoi
2, ume, yayoi
```
```002.csv
id,first,last
3, yuki, yayoi
4, chi, yayoi
```

ボリュームに001.csvをアップロードします。
![Screenshot 2024-02-14 at 20.06.43.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/9d8dd117-988f-70eb-de62-42247633b277.png)

少しするとジョブが起動します。
![Screenshot 2024-02-14 at 19.45.08.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/ff7a3a08-dbcf-1dfb-936f-ab97be319d43.png)

ジョブが完了するとテーブルが作成されます。
![Screenshot 2024-02-14 at 20.07.39.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/9edfa7a5-e129-9dcb-ce44-ac708096dc35.png)
![Screenshot 2024-02-14 at 20.07.48.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/8abf9b81-004f-9caf-ae88-a65986bd44ab.png)

次いで、002.csvをアップロードします。すると、再びジョブが起動します。

テーブルを確認すると、002.csvの内容が追記されていることがわかります。
![Screenshot 2024-02-14 at 20.09.46.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/2210daa2-54f9-f978-b9d4-4b5405c11d22.png)

お手軽にファイル処理のパイプラインを起動できるファイル到着トリガー、ご活用ください！

### はじめてのDatabricks

[はじめてのDatabricks](https://qiita.com/taka_yayoi/items/8dc72d083edb879a5e5d)

### Databricks無料トライアル

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