Delta Live Tablesの本格運用

これらの記事の続きです。

https://qiita.com/taka_yayoi/items/a59fe4b02b202e5f390d

https://qiita.com/taka_yayoi/items/d13efe5e0fa9cbe1a1ad

# Delta Live Tablesの本格運用

DLTはパフォーマンスと使いやすさのために自動でデータを最適化します。
![Screenshot 2024-05-09 at 9.23.29.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/eb3851b9-bb8e-abc1-d2cf-5e0997c0870a.png)

**ベストプラクティス**

- What:
    - DLTはDLTのテーブルを作成する際に、Deltaのベストプラクティスを組み込みます。

- How:
    - DLTは以下のプロパティを設定します:
    - optimizeWrite 
    - autoCompact 
    - tuneFileSizesForRewrites 

**物理データ**

- What: 
    - DLTは、コストを最小化し、パフォーマンスを最適化するために、物理データを自動で管理します。

- How:
    - 日次でvacuumを実行
    - 日次でoptimizeを実行
    - どのように最適化するのかを指示することもできます(ZORDER)

**スキーマ進化**

- What:
    - スキーマ進化が動作します。
- How:
    - マテリアライズドビューのカラムの追加/削除/名称変更を行うと自動で適切な動作をします。
    - ストリーミングテーブルでカラムを削除すると、古い値は保持されます。

## ライフサイクル管理(Unity Catalogのみ)

DLTのコードから何かを削除すると、自動で対応するデータを削除します。
![Screenshot 2024-05-09 at 9.26.42.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/11674feb-dd94-1b41-385c-7c274ae88094.png)

- DLTで移行スクリプトが不要に
    - コードからデータセットを削除すると、Unity Catalogから削除します
    - テーブルのプロパティを削除するとクリアします
    - パーティショニングを削除するとそのパーティションを削除します
- パイプラインを削除すると、作成されたすべてのテーブルをクリーンアップします
- 削除は遅延処理かつ記録されるので、データを失うことなしに間違いから復旧できます

## パイプラインUI

ETLのデバッグとオペレーションのワンストップショップを提供します。
![Screenshot 2024-05-09 at 9.27.45.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/fc64b389-0934-766a-1292-af05231b8c27.png)

- テーブル間のデータフローを可視化
- それぞれのテーブルのメタデータと品質を特定
- 過去の更新処理へのアクセス
- オペレーションのコントロール
- イベントのディープダイブ

## パイプラインの設定

- パイプラインモード:
    - トリガーモード - すべてのデータの処理が完了したらすぐに停止します。
    - 連続モード - 新規データを連続的に処理します。
- クラスターモード:
    - 固定サイズ - N個のノードを割り当て
    - 強化オートスケール - 負荷に応じてクラスターをスケールアップ/ダウン
    - レガシーオートスケール - バッチでのみ動作しますが、ストリームでは動作しません
- エディション - core、pro、advanced ([doc](https://www.databricks.com/jp/product/pricing/delta-live))
- チャネル - current あるいは preview (主にテスト用)
- ターゲットスキーマ - テーブルを作成するデータベース名
- ストレージ - クラウドロケーション (ベストプラクティス: DBFSルートは使わないでください！)
- クラスター設定をカスタマイズできます。

# オーケストレーション

## パイプラインの起動

Delta Live Tablesでは結果の更新をどの頻度で行うのかを選択できます。
![Screenshot 2024-05-09 at 9.29.50.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/23dad9b9-d6aa-7bef-b712-819fe931b71b.png)

- Triggered: 手動
- Triggered: Databricksジョブでスケジュール
- 連続

## ワークフローによるオーケストレーション

ワークフローを用いてTriggered DLTパイプラインをオーケストレーションできます。
![Screenshot 2024-05-09 at 9.32.21.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/67bfd10f-e2e7-cfa3-d469-ae84a7a72642.png)

- マルチタスクジョブでDLTパイプラインやその他のタスクを容易にオーケストレートするためにDatabricksジョブを活用します
- Databricksプラットフォームと完全にインテグレーションされており、結果の調査、デバッグを迅速にします
- マルチクラウド環境のワークロードをオーケストレート、管理します
- また、Apache AirflowやAzure Data Factoryのデータ処理ワークフローの一部としてDelta Live Tablesパイプラインを実行できます

## ベストプラクティス：オートスケーリングでコスト削減

オートスケーリングで予算を設定してDLTにチューニングをお任せください。
![Screenshot 2024-05-09 at 9.36.34.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/79addb9a-ebd6-fb7f-a99a-38255a4c4231.png)

プロダクションでの様々な設定に対する考え方:
- 最小ノード数: デフォルトのまま
- 最大ノード数: タイムリーな結果に対して支払いたいと考える最大値
- インスタンスタイプ: 設定しない – DLTが適切なインスタンスタイプを選択できる自由度を確保

この戦略はクラウドの弾力性を活用します:
- スピードアップできると判断した場合にのみ、オートスケーリングはリソースを追加します。
- 不要になるとすぐにリソースは解放されます。すべてのアップデートが完了するとクラスターはシャットダウンします。
- SLAを違反しないように、最大のサイズを大きするほど、入力ボリュームの変化に対して堅牢になります。

# デプロイメント

## DLTパイプラインのデプロイメント

- DLTパイプライン = ノートブック/ファイル + パイプライン設定
- DLTパイプラインは以下の方法でデプロイ/変更/削除できます:
    - [Databricks DLT REST API](https://docs.databricks.com/api/workspace/pipelines):
        - 課題: ノートブック/パスが変更した際に環境間でのプロモートが困難に
        - 課題: クラスターポリシーなどに対する依存関係の環境間での取り扱いが困難に
    - [Databricks CLIのpipelineコマンド](https://docs.databricks.com/ja/dev-tools/cli/index.html):
        - 課題: 上と同じ
    - Databricks Terraform providerの[databricks_pipeline](https://registry.terraform.io/providers/databricks/databricks/latest/docs/resources/pipeline)リソース
        - メリット: オブジェクト間の依存関係の取り扱いが容易 - ノートブック、クラスターポリシーなど…
        - 課題: [Databricks Terraform exporter](https://registry.terraform.io/providers/databricks/databricks/latest/docs/guides/experimental-exporter)が手動でパイプラインを記述することを回避する助けになるかもしれませんが、Terraformへの習熟が必要
    - おすすめ: [Databricks Asset Bundles](https://learn.microsoft.com/ja-jp/azure/databricks/dev-tools/bundles/)!
        - Databricks CLIの新たなパーツ
        - Databricks Visual Studio拡張とインテグレーション

## TerraformによるDLTパイプラインの配備

![Screenshot 2024-05-09 at 9.39.53.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/c7d1cfef-e74e-7cf0-ea5b-21c3b1a30302.png)

より複雑なセットアップは[こちら](https://github.com/alexott/dlt-files-in-repos-demo/tree/main/terraform/azuredevops)

# CI/CDの自動化: Databricks Asset Bundles(DAB)

## リソース作成をDABが自動化

コードをワークスペースに自動で同期し、ジョブやパイプラインを作成します。
![Screenshot 2024-05-09 at 9.42.26.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/3ad3e66a-bc87-8e57-3a03-7f0c145f595b.png)

- 個別のチェックアウトの作成 → パイプラインは面倒で / エラーが混入しやすくなります
- DAB によってすべてのバージョン管理が可能になります
- 複数環境の作成を自動化します

## yamlでリソースを指定 

ジョブやパイプラインの設定はコードとともにバージョン管理されます。
![Screenshot 2024-05-09 at 9.42.59.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/859e785d-4ac7-a774-3d51-683a1f19fd6b.png)

## DLTにおけるCI/CDパターン

デプロイメントとテストを改善するためにパイプラインを構造化しましょう。

- ソース & 変換処理のノートブック/ファイルの分離
    - ソースとは外部入力から読み込みを行うクエリーです。理想的には、これらには変換処理はなく、非常にシンプルなものです。外部ストレージコストを避けるためにビューを使うこともできます。
    - 変換処理はソースから読み込みを行い実際の処理を行います。
- 検証でエクスペクテーションを活用
    - 理想的には、検証のほとんどは開発とプロダクションの両方で動作します。
    - テスト専用のノートブックには、dev/testのみの追加テストを持つMVを含めることができます。

## ソースと変換処理の分離

dev & testのための小規模なデータや理不尽な入力で置き換えます。
![Screenshot 2024-05-09 at 9.44.05.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/1817e46c-044f-011c-ccf4-a66070a69cc7.png)

## テストデータの作成

それぞれのユースケースでベストな手法を選択します。
![Screenshot 2024-05-09 at 9.44.30.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/cac2b984-47c0-8d17-d3c2-84f46580e875.png)

**プロダクションデータのサブセット**

```sql
CREATE LIVE VIEW sales
AS SELECT * FROM prod.sales
WHERE date > now() – INTERVAL 1 day
```

**ハードコード**

```sql
CREATE LIVE VIEW sales AS 
SELECT -1 AS profit, “test-negative” AS region
UNION ALL
SELECT 99999 AS profit, “test-large” AS region
```

**ファイルから読み込み**

```sql
CREATE LIVE VIEW sales
AS SELECT * FROM json.`/test-data/sales.json`
```

**ランダムに生成**

```py
@dlt.table
def sales():
  return (spark.range(1, 1000)
    .select(random_profit(), random_region()))
```

## バンドルによるソースと変換処理

環境ごとにソースを上書きして使いまわせるようにします。
![Screenshot 2024-05-09 at 9.57.00.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/1a3bc5c9-422e-7f78-3f6f-367cee56a175.png)

# パイプラインの監視

## イベントログ

[イベントログ](https://docs.databricks.com/ja/delta-live-tables/observability.html)は自動ですべてのパイプラインオペレーションを記録します。

- オペレーションの統計情報
    - すべてのオペレーションの時刻と現在のステータス
    - パイプラインとクラスターの設定
    - 行数
- 来歴
    - テーブルのスキーマ、定義、宣言されたプロパティ
    - テーブルレベルのリネージ
    - テーブル更新に用いられたクエリープラン
- データ品質
    - エクスペクテーションの合格 / 失敗 / 削除の統計
    - エクスペクテーションの失敗を引き起こした入力/出力の行数

## ベストプラクティス: イベントログの連携

お使いの既存オペレーションツールでイベントログの情報を活用しましょう。イベントログは単にそれぞれのパイプラインで作成されるDeltaテーブルですので、他のシステムから容易に読み込むことができます。
![Screenshot 2024-05-09 at 9.59.15.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/fb3f0e7b-e979-a97f-290f-e6be49d6d605.png)

## イベントログのクエリーの例

以下からの引用です。

https://www.databricks.com/blog/2022/12/08/build-reliable-and-cost-effective-streaming-data-pipelines.html

```sql
SELECT timestamp, details :cluster_resources.num_executors, details :cluster_resources.optimal_num_executors,
details :cluster_resources.latest_requested_num_executors, details :cluster_resources.avg_num_queued_tasks
FROM delta.`${event_log.location}`
WHERE event_type = 'cluster_resources' AND origin.update_id like '${latest_update.id}'
```
![Screenshot 2024-05-09 at 10.00.50.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/11b5cd7f-60ae-1c4c-bac9-dece4cf2761a.png)

## その他のイベントログのクエリー例

**テーブル/エクスペクテーションごとの合格 & 失敗レコード数:**

```sql
SELECT row_expectations.dataset as dataset, row_expectations.name as expectation,
 SUM(row_expectations.passed_records) as passing_records, SUM(row_expectations.failed_records) as failing_records
FROM (SELECT explode(from_json(details :flow_progress :data_quality :expectations,
         "array<struct<name: string, dataset: string, passed_records: int, failed_records: int>>")) row_expectations
   FROM delta.`${event_log.location}`
   WHERE event_type = 'flow_progress' AND origin.update_id like '${latest_update.id}')
GROUP BY row_expectations.dataset, row_expectations.name
ORDER BY row_expectations.dataset, row_expectations.name
```

**オートスケールのイベント:**

```sql
SELECT timestamp, details :autoscale.status,
 Double(case when details :autoscale.status = 'RESIZING' then details :autoscale.requested_num_executors else null end) as starting_num_executors,
 Double(case when details :autoscale.status = 'SUCCEEDED' then details :autoscale.requested_num_executors else null end) as succeeded_num_executors,
 Double(case when details :autoscale.status = 'PARTIALLY_SUCCEEDED' then details :autoscale.requested_num_executors
     else null end) as partially_succeeded_num_executors,
 Double(case when details :autoscale.status = 'FAILED' then details :autoscale.requested_num_executors else null end) as failed_num_executors
FROM delta.`${event_log.location}` WHERE event_type = 'autoscale' AND origin.update_id like '${latest_update.id}' ORDER BY timestamp DESC
```
# 障害対応

DLTはビルトインのリトライロジックで一時的な問題に対応します。

- DLTでは、スピードと信頼性をバランスを取るためにリトライのエスカレーションを行います
    - 個々のトランザクションをリトライします。
    - 次に、不正な状態にある場合にはクラスターを再起動します。
    - 最後の処理の成功以降にDBRがアップグレードされた場合には、退行を検知するために自動で古いバージョンで試行します。
- トランザクション保証
    - テーブルに対するオペレーションは原子的です。
    - パイプラインのそれぞれのテーブルのアップデートは原子的ではありません。ベストエフォートで可能な限り多くのテーブルをアップデートします。

# ベストプラクティス：すべてのテーブル属性をコードとして管理

ガバナンス、ディスカバリー、データレイアウトに対する単一のソースを実現します。

```sql
CREATE MATERIALIZED VIEW report(
  customer_id LONG COMMENT "the customer id in salesforce", …
)
TBLPROPERTIES (
  delta.deletedFileRetentionDuration = "interval 30 days",
  pipelines.autoOptimize.zOrderCols = "timestamp,user_id",
  my_etl.quality = "gold",
  my_etl.has_pii = "false"
)
COMMENT "Weekly E-Staff report on key customer metrics, produced by growth team."
AS …
```

- **カラム**と**テーブル全体**のコメントは、データのソースやどのような変換処理が行われているのかに関する**ドキュメント**作成に活用できます。
- `delta.*`プロパティは格納されるデータの**保持期間**やその他の物理観点でDelta Lakeによってサポートされるオプションです。
- `pipelines.*`プロパティはデータをどのように整理するのかを含むDLTのオペレーションの制御に関するオプションです。
- **アプリケーション固有のプロパティ**はキー/バリューペアとして格納されます。
- ベストプラクティスは、これらのプロパティを特定するために**共通のプレフィクス**を使うと言うものです。

### はじめてのDatabricks

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

### Databricks無料トライアル

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