Delta Live Tablesを使い始めると感じる疑問(回答付き)

本書では[Delta Live Tables(DLT)](https://databricks.com/jp/product/delta-live-tables)を使い始めると疑問に感じるポイントと、それに対する回答をまとめたものです。

本書でカバーされている以外の質疑に関しては、[Delta Live TablesのFAQ](https://qiita.com/taka_yayoi/items/ab79b80936f8e87403d6)を参照ください。

:::note info
Delta Live Tablesを利用するには申請が必要です。申請に関してはDatabricks担当者にお問い合わせください。
:::

# Delta Live Tablesのご紹介 

Delta Live Tablesは、新鮮かつ高品質なデータを提供するデータパイプラインを容易に構築、管理できる機能を提供します。

- **パイプラインの容易な開発、維持**
データパイプライン(バッチ、ストリーミング)を構築、管理するための記述ツール

- **自動テスト** 
ビルトインの品質管理、データ品質モニタリング

- **簡素化されたオペレーション** 
パイプラインオペレーションに対するディープな可視化を通じた自動エラーハンドリング

## データパイプラインの容易な構築、維持

![スクリーンショット 2022-03-07 18.13.43.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/b756b895-177f-94d7-f127-95f023d9083a.png)

- ビジネスロジックとテーブルの依存関係をまとめて、**宣言的にデータパイプラインを構築**します。
- 構造化/非構造化データを**バッチ、ストリーミング**で実行します。
- 複数の環境(テスト/ステージング/プロダクション)で**ETLパイプラインを再利用**できます

## データに対する信頼性の確保

- Deltaのエクスペクテーション(データに対する期待をルールとして定義)により**不正なデータがテーブルに流れ込むことを防ぎます**。
- 事前定義されたエラーポリシー(失敗、欠損、警告、データ検疫)によって**データ品質エラーを回避、対策**します。
- 長期にわたる**データ品質のトレンドをモニタリング**します。

## 信頼性とオペレーションのシンプルさを維持しながらスケール

![スクリーンショット 2022-03-07 18.18.09.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/031e512b-a41a-17c2-de67-49cb53490df0.png)


- オペレーションの状態とデータリネージュをビジュアルで追跡できるツールを用いることで、**パイプラインオペレーションを深く理解**することができます。
- 自動エラーハンドリングおよび容易なリトライにより**ダウンタイムを削減**します。
- シングルクリックによるデプロイメント、アップグレードにより**メンテナンスをスピードアップ**します。

# Delta Live Tablesに対する疑問及び回答

## Delta Live Tablesを使うことのメリットは？Databricksのジョブでいいのでは？

- 非常に高い柔軟性を必要するのであればジョブでもいいですが、DLTを使うと以下のメリットがあります。
    - ETLイベントログが自動で記録されます。ジョブの場合、Pythonなどでロギングのロジックを組む必要があります。
    - 宣言型でETLロジックを組める(最初のテーブルはこうである、次のテーブルはこうである)ので処理の見通しがよくなります。PySparkやSQLで全てのロジックを記述することはできますが、見通しを確保するには職人芸が必要です。
    - ETL処理をモジュール化できるというメリットもあると思います。

:::note
**参考資料**
 [Delta Live Tablesでインテリジェントデータパイプラインを実装する5つのステップ - Qiita](https://qiita.com/taka_yayoi/items/8466ceb6b689541327fc)
:::

## Delta Live Tablesはどのように動作するのですか？

- 内部で動いているのはSparkの構造化ストリーミングとDelta Lakeです。
- DLTパイプラインの定義に基づいて構造化ストリーミングが動作し、Delta Lakeに中間データ、最終データ、イベントログが書き込まれます。これによって、パイプラインの途中でエラーが起きてもパイプライン全体のロールバックが可能となっています。ストリーミング処理のチェックポイントも生成されます。
- このため、入力データのスキーマ変更の影響範囲も構造化ストリーミングの仕様に依存します。

:::note
**参考資料**
 [Databricksにおける構造化ストリーミングの本格運用 - Qiita](https://qiita.com/taka_yayoi/items/6a563259261f42d24b2f)
:::

## IncrementalテーブルとCompleteテーブルの違いは？

- 実行の都度、総洗い替えになるのがComplete、差分のみを処理するのがIncrementalです。すべてをCompleteテーブルで構成してバッチで実行するのがDLTで一番シンプルな構成になります。
- Incrementalテーブルがある場合でも、全てのデータを再処理する際にはFull Refreshを実行します。

:::note
**参考資料**
[Delta Live Tablesユーザーガイド - Qiita](https://qiita.com/taka_yayoi/items/6726ad1edfa92d5cd0e9#%E3%82%A4%E3%83%B3%E3%82%AF%E3%83%AA%E3%83%A1%E3%83%B3%E3%82%BF%E3%83%AB%E3%81%AA%E3%83%87%E3%83%BC%E3%82%BF%E5%87%A6%E7%90%86)
:::

## Delta Live Tablseでスキーマエボリューションはサポートされていますか？

はい。Auto Loaderが読み込むファイルのスキーマが変更された場合、下流に変化が伝播します。

:::note
**参考資料**
[DatabricksのAuto Loader \- Qiita](https://qiita.com/taka_yayoi/items/df143647dcf5942b51c6)
:::

## パイプラインのロジックを複数のノートブックに分割できますか？

できます。パイプラインのlibrariesでノートブックを複数指定できます。ノートブックの順序は問いません。

:::note
**参考資料**
[Delta Live Tablesの設定 - Qiita](https://qiita.com/taka_yayoi/items/834b56b30849d63b52d0#%E3%83%91%E3%82%A4%E3%83%97%E3%83%A9%E3%82%A4%E3%83%B3%E3%81%AE%E8%A4%87%E6%95%B0%E3%81%AE%E3%83%8E%E3%83%BC%E3%83%88%E3%83%96%E3%83%83%E3%82%AF%E3%81%AE%E8%A8%AD%E5%AE%9A)
:::

## パイプライン内でロジックを共有できますか？

UDF(ユーザー定義関数)を使います。UDFで定義されたロジックは他のDLTノートブックからも使用できます。エクスペクテーションからもUDFを呼び出せるので、柔軟にロジックを記述することができます。

:::note
**参考資料**
[Delta Live TablesでUDFを活用する - Qiita](https://qiita.com/taka_yayoi/items/123347aa8323b4a90730)
:::

## 複数のノートブック、複数のパイプラインで変数や関数を共有できますか？

ReposのFiles機能を使います。Reposに登録された.pyファイルはDLTパイプラインノートブックから参照することができます。

:::note
**参考資料**
[Delta Live Tables(DLT)でロジックをpyファイルにモジュール化して共通化する - Qiita](https://qiita.com/taka_yayoi/items/59d4cb27d38ba61d8c9b)
:::

## DLTの処理をスケジューリングできますか？

ジョブのタスクでパイプラインを指定してください。

:::note
**参考資料**
[Databricks JobsからDelta Live Tablesパイプラインを呼び出す - Qiita](https://qiita.com/taka_yayoi/items/ec547debcb76f7eadfe5)
:::

## DLTのログはどのように活用すればいいのですか？

Delta形式で保存されるイベントログはそのままでも参照できますが、詳細情報が含まれている`detail`列にはJSON文字列が入っているのでパーシングが必要です。

以下のサンプルを参考にしてください。

```scala:Scala
var json_parsed = spark.read.json(spark.table("<DLTログテーブル>").select("details").as[String])
var json_schema = json_parsed.schema
```

```scala:Scala
import org.apache.spark.sql.functions._

val parsed_event_log = spark.table("dlt_takaaki_yayoi_databricks_com_db.event_log_raw").withColumn("details_parsed", from_json($"details", json_schema))

parsed_event_log.write.format("delta").option("optimizeWrite", "true").saveAsTable("<パース済みDLTログテーブル>")
```

```sql:SQL
select *
from <パース済みDLTログテーブル>
```

![スクリーンショット 2022-03-07 18.34.07.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/98bf41cb-07e1-a509-73b5-1deb89d1fd65.png)

```sql:SQL
SELECT
  details:flow_definition.output_dataset,
  details:flow_definition.input_datasets,
  details:flow_definition.flow_type,
  details:flow_definition.schema,
  details:flow_definition.explain_text,
  details:flow_definition
FROM <パース済みDLTログテーブル>
WHERE details:flow_definition IS NOT NULL
ORDER BY timestamp
```

![スクリーンショット 2022-03-07 18.35.15.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/9dffac3f-1db5-266e-8dd2-a0475e67deb6.png)

以下の様に、入力のデータセット、出力のデータセットのペアから構成されるリネージュを取得することができます。

```sql:SQL
SELECT
  details:flow_definition.input_datasets,
  details:flow_definition.output_dataset
FROM dlt_takaaki_yayoi_databricks_com_db.event_log
WHERE details:flow_definition IS NOT NULL
```

![スクリーンショット 2022-03-07 18.36.08.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/51195b4a-a098-cf35-0101-e56f1a91487d.png)


## DLTのクラスターはどのように動作しますか？

- DLTのクラスターはDLT向けにカスタマイズされたクラスターです。クラスターライブラリは使用できないので、ライブラリをインストールするにはDLTノートブック内で`%pip`してください。
- クラスター自動停止の挙動はパイプラインのモード(Development/Production)で変わるので注意してください。Productionの場合Jobsクラスターと同じ挙動(ジョブが終了するとすぐに停止)となりますが、Developmentの場合は2時間のアイドル状態の後に停止されます。なので、ジョブにDLTを組み込む際には、必ずProductionモードになっていることを確認してください。


### Databricks 無料トライアル

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