Delta Live Tablesのコアコンセプト

こちらの前段となります。

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

https://qiita.com/taka_yayoi/items/03554306b8b83198b421

# レイクハウスの基盤は良いデータです

すべてのデータ専門家はクリーンで新鮮、信頼できるデータを必要とします。
![Screenshot 2024-05-03 at 10.06.45.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/369193ba-d674-dcd1-0c44-c1d1f154216c.png)

# しかし、現実はそんなに単純ではありません

大規模データの品質と信頼性の維持は複雑で不安定です。
![Screenshot 2024-05-03 at 10.07.22.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/bbe3f363-e259-56fa-6b23-e063250a97dd.png)

# データ専門家としての人生…

CEOから無茶振りされることもあります。
![Screenshot 2024-05-03 at 10.08.04.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/359f278f-79f0-e7d7-5aad-cff52b6e102b.png)

# クエリーからプロダクションへ

SQLクエリーを本番システムに持っていくには多くの苦労が伴います。
![Screenshot 2024-05-03 at 10.08.35.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/7a20f823-191f-b4f8-5e3b-e4c2953bbfc1.png)

SQLクエリー以外にも、以下のようなことを検討、実装しなくてはなりません。

- バージョン管理
- インフラのデプロイ
- 品質チェック
- ガバナンス
- データの検知
- バックフィル対応
- 依存関係の管理
- 日次のパーティション計算
- チェックポイント & リトライ

![Screenshot 2024-05-03 at 10.09.28.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/771ff659-2dff-de12-dcc0-b8cd1f872349.png)

データエンジニアは、変換処理ではなくツールの準備に時間が費やされます。
![Screenshot 2024-05-03 at 10.11.08.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/53a2b56d-e72a-99e0-dfcb-01e793822b4a.png)

フォーカスしなくてはいけないのは、**データから価値を得る**ことです。
![Screenshot 2024-05-03 at 10.11.31.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/9e19e4b6-9c25-3b57-c485-090c403d2528.png)

# Delta Live Tablesのご紹介

LIVEを追加するだけでクエリーからプロダクションパイプラインへ。
![Screenshot 2024-05-03 at 10.12.12.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/98463285-1c55-a023-9db3-b19870cf93a9.png)

# Delta Live Tablesの基本コンセプト

## 宣言型プログラミングのご紹介

宣言型プログラムは**どのようにするのか**ではなく**何をすべきか**を記述します。
![Screenshot 2024-05-03 at 10.12.56.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/a13b8c67-fe76-4c24-6426-cb818bf97fb8.png)

**命令型プログラム**

```py
numbers = […]
sum = 0
for n in numbers:
  sum += n
print(n)
```

**宣言型プログラム**

```sql
SELECT sum(n) FROM numbers
```

![Screenshot 2024-05-03 at 10.13.41.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/0cfa7576-e1e3-8468-f137-86c9d7036d19.png)

## ワークフローかDLTか？

大抵両方です: DLTを含むすべてをワークフローはオーケストレートできます。

- **すべてのタスクの実行**に**ワークフロー**を使用
    - スケジュールで実行
    - 他のタスクが完了したら実行
    - ファイルが到着したら実行
    - 他のテーブルが更新された実行

- **データフロー**の管理に**DLT**を使用
    - Deltaテーブルの作成/更新
    - 構造化ストリーミングの実行

## なぜ宣言型プログラミング？

**あなたの目的を理解**することでDLTは**つまらないことの面倒を見てくれます**。

- 定型文の排除
    - 耐障害性
    - 状態管理
    - オブジェクトのライフサイクル
    - スケジュール、依存関係、並列度
    - 一般的なデータの問題 (CDC、スキーマ進化)
- 最適化
    - 動的実行や複数クエリーに対するベストな戦略を選択
- オペレーション
    - HMR(Hot Module Replacement) – システムの退行やクラウド問題を自動で検知、対策

## DLTのコアの抽象化

**データセット**を定義すると、DLTは自動でそれらを**最新の状態**に維持します。

![Screenshot 2024-05-03 at 10.18.45.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/64925204-110a-79a0-c3e2-adc9c32c597f.png)

### ストリーミングテーブル

**(複数の)ストリーム**が書き込まれるDeltaテーブル

用途: 
- データの取り込み
- 低レーテンシーの変換処理
- 大規模データ

### マテリアライズドビュー

Deltaテーブルに格納される**クエリーの結果**

用途: 
- データの変換
- 集計テーブルの構築
- BIのクエリーとレポートの高速化

## ストリーミングテーブルとは？

構造化ストリーミングが書き込みを行うDeltaテーブルです。
![Screenshot 2024-05-03 at 10.20.15.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/c9ce2c3e-78a5-c6b0-0466-c8834e581a71.png)

```sql
CREATE STREAMING TABLE report
AS SELECT *
FROM read_files("/mydata/", "json")
```

- ストリーミングテーブルはKafka、Kinesis、Auto Loader(クラウドストレージ上のファイル)のような追加オンリーのデータソースから読み込みを行います
- ストリーミングテーブルは、それぞれの入力レコードを一度だけ読み込むことで、コストとレーテンシーを削減します
- ストリーミングテーブルはアドホックなデータ操作(GDPR等)のためのDML(UPDATE, DELETE, MERGE)をサポートします

## マテリアライズドビューとは？

事前に計算され、Deltaに格納されるクエリーの結果です。
![Screenshot 2024-05-03 at 10.21.15.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/720b0b92-3173-a854-11e2-249a78f090b4.png)


```sql
CREATE MATERIALIZED VIEW report
AS SELECT sum(profit)
FROM prod.sales
GROUP BY date
```

- マテリアライズドビューは、最後にアップデートされた際の定義済みクエリーの結果を常に返却します(スナップショット)
- マテリアライズドビューのデータを変更することはできません。クエリーを変更することはできます。

## Pythonの活用

高度なデータフレームのコードとUDFの記述が可能です。
![Screenshot 2024-05-03 at 10.22.40.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/c1594e84-186b-9ec2-689f-10f3b587b36d.png)

```py
import dlt

@dlt.table
def report():
  df = spark.table("LIVE.events")
  return df.select(…)
```
```py
spark.udf.register(
  "complex_function",
   lambda x: …)
```

- データフレームを返却する任意の関数に`@dlt.table`を追加します。
- 単一のパイプラインでSQL/Pythonのノートブックを混成、マッチさせます。
- しかし、単一のノートブックではすべてPythonかすべてSQLである必要があります(ノートブックレベルで混成、マッチはできません)
- PySpark、Koalas、SQL文字列を用いてデータフレームを構成できます。

:::note info
**クイックノート: 用語の進化**
- STREAMING LIVE TABLE ➡ STREAMING TABLE
- LIVE TABLE ➡ MATERIALIZED VIEW

セマンティクスは同じで、古い文法との互換性をサポートします。我々のゴールは、文法をシンプルにして他のシステムとマッチさせることです。
:::

## エクスペクテーションによる適正性の保証

エクスペクテーションはプロダクションのデータ品質を保証するテストです。

```sql
CONSTRAINT valid_timestamp
EXPECT (timestamp > '2012-01-01’)
ON VIOLATION DROP
```
```py
@dlt.expect_or_drop(
  "valid_timestamp",
  col("timestamp") > '2012-01-01')
```

エクスペクテーションは処理の過程でそれぞれの行の検証に用いられるtrue/falseのエクスプレッションです。

DLTはエクスペクテーションの違反レコードに対する柔軟なポリシーを提供します:
- 不正レコード数の追跡
- 不正レコードの削除
- 一つの不正レコードで処理を中止
- 検疫 (提供予定、あるいは[クックブックのレシピ](https://docs.databricks.com/ja/delta-live-tables/expectations.html#quarantine-invalid-data)をお使いください)

## パイプラインの作成

DatabricksのUIによるパイプラインの作成方法。
![Screenshot 2024-05-03 at 10.24.57.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/1b3c6c77-c64a-01ee-7b81-e29cc7d9fc27.png)

1. CREATE ST/MVの記述
    - ノートブックでテーブル定義を記述します。
    - Databricks Reposによってテーブル定義のバージョン管理が可能になります。
1. パイプラインの作成
    - 一つのパイプラインでは、一つ以上のノートブックと必要な設定を組み合わせます。
1. スタートをクリック
    - DLTはパイプラインのすべてのテーブルを作成、更新します。

## DLTパイプラインへのデータの取り込み

- すべてのビルトインのSparkデータソース
    - ファイル - バッチとストリーミングの両方、しかし、ファイル一覧のみ、スキーマ進化は未対応
    - JDBC - DBRバンドルのデータベースドライバーのみ、バッチのみ
- Databricks Auto Loader
    - ストリーミングテーブルのみ
    - ファイル通知や最適化ファイル一覧モード
    - スキーマ進化！(DLTではデフォルト)
    - Python & SQL APIの両方
- ストリーミングシステム (多くの場合Python):
    - Kafka ([read_kafka](https://docs.databricks.com/en/sql/language-manual/functions/read_kafka.html)関数でSQLでもサポート)
    - Kinesis
    - EventHubs (Kafkaコネクター経由)
- より多くのソースを提供予定…

## データロードの例

**SQLを用いたバッチモードでJSONファイルを読み込み**

```sql
CREATE MATERIALIZED VIEW taxi_raw
AS SELECT * FROM json.`/.../json/`
```

**Pythonを用いたストリーミングモードでParquetファイルを読み込み**

```py
@dlt.table
def raw_parquet():
 return spark.readStream.format("parquet").load(raw_path)
```

**[Kafkaコネクターを用いたAzure Event Hubs](https://alexott.blogspot.com/2022/06/delta-live-tables-recipes-consuming.html)からの読み込み**

```py
@dlt.table(table_properties = {"pipelines.reset.allowed": "false"})
def bronze():
 eh_sasl = f'kafkashaded.org.apache.kafka.common.security.plain.PlainLoginModule required username="$ConnectionString" password="{readConnectionString}";'
 kafka_options = { "kafka.bootstrap.servers": f"{eh_namespace}.servicebus.windows.net:9093",
                "kafka.sasl.mechanism": "PLAIN", "kafka.security.protocol": "SASL_SSL",
                "startingOffsets": "earliest", "failOnDataLoss": False,
                "kafka.sasl.jaas.config": eh_sasl, "subscribe": eh_topic, }
 df = spark.readStream\
   .format("kafka")\
   .options(**kafka_options)\
   .load()\
   .withColumn("value", F.col("value").cast("string"))
```

## [Databricks Auto LoaderとDLT](https://docs.databricks.com/ja/sql/load-data-streaming-table.html)の活用

### SQLの[read_files](https://docs.databricks.com/en/sql/language-manual/functions/read_files.html)関数を使用(以前はcloud_files)

```sql
CREATE STREAMING TABLE raw
AS SELECT *
 FROM read_files('/path/…/*.csv', 'csv', 
    map('header', 'true', 
    'cloudFiles.inferColumnTypes', 'true') 
  );
```

パラメーター:
- ファイルパス (必須)
- ファイルフォーマット (必須)
- 設定 (`map<string, string>`, オプション)

### Pythonを使用

```py
@dlt.table
def raw():
 return (
   spark.readStream.format("cloudFiles")
     .option("cloudFiles.format", "csv")
     .option("cloudFiles.inferColumnTypes", "true")
     .load("/path/…/*.csv")
```

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

- 可能であれば[ファイル一覧ではなくファイル通知モード](https://docs.databricks.com/ja/ingestion/auto-loader/file-detection-modes.html)を使いましょう - ファイル一覧の優れたパフォーマンス、少ないオーバーヘッド。
- [スキーマ推定](https://docs.databricks.com/ja/ingestion/auto-loader/schema.html)を使いましょう。DLTでAuto Loaderを使う際にはデフォルトでスキーマ進化が有効化されます！
- モニタリングに関しては、[プロダクションのベストプラクティス](https://docs.databricks.com/ja/ingestion/auto-loader/production.html)に従いましょう。

## 無制限の保管のためにDeltaを活用(リファレンスアーキテクチャ)

Deltaは一時的なソースに対して、安価で柔軟、管理可能なストレージを提供します。
![Screenshot 2024-05-03 at 10.31.41.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/f6e5ff2c-c8df-c947-3344-88baf61f098b.png)

- ストリーミングソースでは、コンプライアンスリスクを回避し、コストを削減するために短期間の保持期間を使用します。
- ブロンズテーブルではバグや重要なデータの削除を引き起こす可能性のある複雑な変換処理を回避します。
- ブロンズテーブルでは、履歴を保持することで、GDPRやその他のコンプライアンスタスクの実行が容易になります。
- `pipelines.reset.allowed=false`を設定することで、後段の処理でデータを失わないフルリフレッシュを確実にします。

[Delta Live Tables開発のベストプラクティス](https://qiita.com/taka_yayoi/items/d13efe5e0fa9cbe1a1ad)や[Delta Live Tablesの本格運用](https://qiita.com/taka_yayoi/items/03554306b8b83198b421)に続きます。

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

https://qiita.com/taka_yayoi/items/03554306b8b83198b421

### はじめてのDatabricks

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

### Databricks無料トライアル

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