Databricks Delta Live Tablesでチェンジデータキャプチャをシンプルに

[Change Data Capture With Delta Live Tables \- The Databricks Blog](https://www.databricks.com/blog/2022/04/25/simplifying-change-data-capture-with-databricks-delta-live-tables.html)の翻訳です。

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

このガイドでは、データレイクにおける新規レコードを特定し、データへの変更を補足するために、どのようにDelta Live Tablesパイプラインのチェンジデータキャプチャを活用できるのかを説明します。Delta Live Tablesを用いることで、スケーラブル、高信頼、低レーテンシーのデータパイプラインを構築できることに加え、最低限の計算リソースとシームレスな遅延データハンドリングを用いたデータレイクに対するチェンジデータキャプチャを実行することができます。

:::note
**注意**
Delta Live Tables (DLT)と宣言型ETL定義にを用いたスケーラブルかつ高信頼なパイプラインの作成方法を説明している[Delta Live Tablesを使ってみる](https://qiita.com/taka_yayoi/items/0cde1f2732ff859a726f)を試すことをお勧めします。
:::

# チェンジデータキャプチャの背景

チェンジデータキャプチャ([CDC](https://en.wikipedia.org/wiki/Change_data_capture))は、ニアリアルタイムのデータアプリケーションにおける顧客の追跡、注文・商品のステータスの追跡の様に、データベースにおけるインクリメンタルな変更(データの削除、追加、更新)を特定、補足するプロセスです。CDCでは、新規イベントが生じるのに合わせて継続的かつインクリメンタルな方法でデータを処理することで、リアルタイムのデータ進化を実現しています。[2025年までには企業の80%がマルチクラウドの戦略を実施](https://solutionsreview.com/data-integration/whats-changed-2020-gartner-magic-quadrant-for-data-integration-tools/)しようとしていることから、複数の環境におけるETLパイプラインにおけるすべてのデータの変更をリアルタイムかつシームレスに集中管理することが重要となります。

CDCイベントを補足することで、Databricksユーザーは外部システムとデータを結合できつつも、レイクハウスのDeltaテーブルとしてソーステーブルを再マテリアライズでき、分析を実行できる様になります。DatabricksにおけるDelta Lakeの`MERGE INTO`コマンドによって、お客様は自分のデータレイクで効率的にレコードのupsert、deleteを行うことができます。以前ディープダイブした[こちらのトピック](https://qiita.com/taka_yayoi/items/02100199211610a36252)をチェックすることもできます。これは、リアルタイムのビジネスデータでデータレイクを最新の状態に保っている数多くのDatabricksのお客様においては一般的なユースケースとなっています。

Delta LakeはデータレイクにおけるリアルタイムCDC同期に対する完全なソリューションを提供しますが、みなさまのアーキテクチャをさらにシンプルに、さらに効率的、スケーラブルにするDelta Live Tablesにおけるチェンジデータキャプチャを発表できることを嬉しく思っています。DLTによって、ユーザーはSQLやPythonを用いてシームレスにCDCデータを取り込める様になります。

Deltaテーブルによる初期のCDCソリューションでは、ソースデータセットの複数行がターゲットDeltaテーブルに存在する同じ行にマッチし、アップデートしようとした際に処理が失敗することを避けるために、手動でのデータの並び替えを必要とする`MERGE INTO`オペレーションを使用していました。並び順を守らないデータに対応するためには、複数マッチの可能性を排除し、それぞれのキーの最新の変更のみを保持するために`foreachBatch`実装を用いてソーステーブルを前処理する追加のステップが必要でした。DLTにおける新たな`APPLY CHANGES INTO`オペレーションは、手動でのデータエンジニアリングの作業を必要とせず、自動かつシームレスに順序を守らないデータをハンドリングします。

# Databricks Delta Live TablesによるCDC

この記事では、CDCデータが外部システムから到着する一般的なCDCユースケースにおけるDelta Live Tablesパイプラインにおける`APPLY CHANGES INTO`コマンドの使い方をデモします。Debezium、Fivetran、Qlik Replicate、Talend、StreamSetsのように数多くのCDCツールが存在しています。固有の実装は異なりますが、これらのツールは一般的には、データ変更の履歴を補足してログに記録します。後段のアプリケーションがこのログを活用します。我々のサンプルでは、Debezium、FivetranのようなCDCツールからクラウドオブジェクトストレージにデータが到着します。

クラウドオブジェクトストレージにデータを提供するさまざまなCDCツールや、Apache Kafkaのようなメッセージキューからデータが到着します。多くの場合、取り込みに用いられるCDCは我々がメダリオンアーキテクチャというものに流入します。メダリオンアーキテクチャでは、アーキテクチャの各レイヤーを通じて、データの構造と品質を改善していくために用いられるデータデザインパターンです。Delta Live Tablesを用いることで、シームレスにCDCフィードからレイクハウスのテーブルに変更を適用することができます。この機能とメダリオンアーキテクチャを組み話せることで、大規模な分析ワークロードを通じて容易に変更をインクリメンタルに適用できる様になります。CDCとメダリオンアーキテクチャを組み合わせて活用することで、変更データ、追加データのみを処理すればよいので、ユーザーに対して複数のメリットを提供します。つまり、ユーザーは最新のビジネスデータを用いて、ゴールドテーブルをコスト効率高く最新の状態に保つことができます。

:::note
**注意**
ここでのサンプルはCDCのSQLパージョン、Pythonバージョンの両方に適用でき、バリエーションの評価、オペレーション固有の方法にも適用することができます。[こちら](https://docs.databricks.com/data-engineering/delta-live-tables/delta-live-tables-cdc.html#python)の公式ドキュメントをご覧ください。
:::

## 前提条件

このガイドの大部分を実行するには、以下の点に慣れ親しんでいる必要があります。

- SQLあるいはPython
- Delta Live Tables
- ビッグデータシステムの操作、ETLパイプラインの開発
- Databricksのインタラクティブノートブック、クラスター
- Databrikcsワークスペースにおける、新規クラスター作成、ジョブ実行、外部クラウドオブジェクトストレージあるいは[DBFS](https://qiita.com/taka_yayoi/items/f15b50469b368907ec3b)にデータを保存する権限。
- 本記事で作成するパイプラインにおいては、データ品質制約の強制をサポートしている`Advanced`製品エディションを選択する必要があります。

## データセット

ここでは、外部データベースからの現実的なデータに模したCDCデータを消費していきます。このパイプラインでは、DebeziumのようなCDCツールが生成するデータセットを生成するために[Faker](https://github.com/joke2k/faker)ライブラリを使用し、Databricksにおける初期取り込みのためにクラウドストレージに持ってきます。[Auto Loader](https://qiita.com/taka_yayoi/items/df143647dcf5942b51c6)を用いて、クラウドオブジェクトストレージからのメッセージをインクリメンタルにロードし、生のメッセージを格納するために、これらをブロンズテーブルに格納します。ブロンズテーブルは、信頼できる唯一の情報源へのクイックなアクセスを実現するためのデータ取り込みを意図したものです。次に、後段のシルバーテーブルに変更を伝播させるためにクレンジングされたブロンズレイヤーテーブルから`APPLY CHANGES INTO`を実行します。シルバーテーブルにデータが流れ込むと一般的には、企業にキーとなるビジネスエンティティのビューを提供する、より洗練され最適化された状態になります。以下の図をご覧ください。
![](https://www.databricks.com/wp-content/uploads/2022/04/db-129-blog-img-2.png)

本記事では、顧客名、メールアドレス、住所、IDと、変更データを説明するための2つのフィールド: オペレーションのコード(DELETE, APPEND, UPDATE, CREATE)を格納するoperation、それぞれのオペレーションのアクションに対応するレコードが到着した日付とタイムスタンプを格納するoperation_dateを持つJSONを必要とするシンプルな例にフォーカスします。

上述のフィールドを持つサンプルデータセットを生成するために、フェイクデータを生成するPythonパッケージであるFakerを使用します。[こちら](https://www.databricks.com/wp-content/uploads/notebooks/DB-129/1-cdc-data-generator.html)からこのデータ生成セクションに関連するノートブックを参照することができます。Databricksの機能であるDBFSを使用します。詳細に関しては[DBFSのドキュメント](https://qiita.com/taka_yayoi/items/f15b50469b368907ec3b)を参照ください。そして、それぞれのフィールドごとに合成データセットを生成し、他のノートブックから合成データセットを参照できる様に、定義されたストレージロケーションに保存するために、PySparkのユーザー定義関数を使います。

## Auto Loaderによる生のデータセットの取り込み

メダリオンアーキテクチャパラダイムによると、ブロンズレイヤーでは最も生のデータ品質を保持します。このステージでは、Auto Loaderを用いてクラウドストレージのロケーションからインクリメンタルに新規データを読み込みます。ここでは、パイプライン設定のconfigurationセクションに生成されたデータセットへのパスを追加し、変数としてソースパスを読み込める様にします。パイプライン設定のconfigurationは以下の様になります。

```json:JSON
"configuration": {
"source": "/tmp/demo/cdc_raw"
}
```

次に、ノートブックからこの設定をロードします。

取り込みを行うブロンズテーブルのSQLバージョンとPythonバージョンを見ていきましょう。

```sql:SQL
SET spark.source;
CREATE STREAMING LIVE TABLE customer_bronze
(
address string,
email string,
id string,
firstname string,
lastname string,
operation string,
operation_date string,
_rescued_data string 
)
TBLPROPERTIES ("quality" = "bronze")
COMMENT "New customer data incrementally ingested from cloud object storage landing zone"
AS 
SELECT * 
FROM cloud_files("${source}/customers", "json", map("cloudFiles.inferColumnTypes", "true"));
```

```py:Python
import dlt
from pyspark.sql.functions import *
from pyspark.sql.types import *

source = spark.conf.get("source")

@dlt.table(name="customer_bronze",
                  comment = "New customer data incrementally ingested from cloud object storage landing zone",
  table_properties={
    "quality": "bronze"
  }
)
def customer_bronze():
  return (
    spark.readStream.format("cloudFiles") \
      .option("cloudFiles.format", "json") \
      .option("cloudFiles.inferColumnTypes", "true") \
      .load(f"{source}/customers")
  )
```

上の文では、JSONファイルから`customer_bronze`というストリーミングLIVEテーブルを作成するためにAuto Loaderを使用しています。Delta Live TablesでAuto Loaderを使う際、DLTによってロケーションは自動で管理されるので、スキーマやチェックポイントのロケーションを指定する必要はありません。

Auto Loaderでは、SQLなら`cloud_files`、Pythonなら`cloudFiles`というパラメーターとしてストレージパスとフォーマットを受け取る構造化ストリーミングソースを提供しています。計算コストを削減するために、低レーテンシーの要件がない場合にはマイクロバッチとして`Triggered`モードでDLTパイプラインを実行することをお勧めします。

## エクスペクテーションと高品質データ

次のステップでは、高品質、多様かつアクセス可能なデータセットを作成するために、`Constraints(制約)`を用いたエクスペクテーションの評価基準による品質チェックを行います。現時点では、制約は保持、削除、処理の失敗を行うことができます。詳細は[こちら](https://qiita.com/taka_yayoi/items/437817339ada5d21b424)を参照ください。合理的な品質モニタリングを行うために、全ての制約は記録されます。

```sql:SQL
CREATE TEMPORARY STREAMING LIVE TABLE customer_bronze_clean_v(
  CONSTRAINT valid_id EXPECT (id IS NOT NULL) ON VIOLATION DROP ROW,
  CONSTRAINT valid_address EXPECT (address IS NOT NULL),
  CONSTRAINT valid_operation EXPECT (operation IS NOT NULL) ON VIOLATION DROP ROW
)
TBLPROPERTIES ("quality" = "silver")
COMMENT "Cleansed bronze customer view (i.e. what will become Silver)"
AS SELECT * 
FROM STREAM(LIVE.customer_bronze);
```

```py:Python
@dlt.view(name="customer_bronze_clean_v",
  comment="Cleansed bronze customer view (i.e. what will become Silver)")

@dlt.expect_or_drop("valid_id", "id IS NOT NULL")
@dlt.expect("valid_address", "address IS NOT NULL")
@dlt.expect_or_drop("valid_operation", "operation IS NOT NULL")

def customer_bronze_clean_v():
  return dlt.read_stream("customer_bronze") \
            .select("address", "email", "id", "firstname", "lastname", "operation", "operation_date", "_rescued_data")
```

## 後段のターゲットテーブルに変更を伝播させるためにAPPLY CHANGES INTO文を使用

`APPLY CHANGES INTO`クエリーを実行する前に、最新のデータを保持したいと考えているターゲットストリーミングテーブルが存在していることを確認する必要があります。存在しない場合には作成する必要があります。以下のセルは、ターゲットストリーミングテーブルのサンプルです。本記事の公開時点では、`APPLY CHANGES INTO`クエリーではターゲットストリーミングテーブルの作成が必要であり、両方ともパイプラインに存在する必要があることに注意してください。そうでない場合にはテーブル作成のクエリーは失敗します。

```sql:SQL
CREATE STREAMING LIVE TABLE customer_silver
TBLPROPERTIES ("quality" = "silver")
COMMENT "Clean, merged customers";
```

```py:Python
dlt.create_target_table(name="customer_silver",
  comment="Clean, merged customers",
  table_properties={
    "quality": "silver"
  }
)
```

これでターゲットのストリーミングテーブルが作成されたので、`APPLY CHANGES INTO`クエリーを用いて後段のターゲットテーブルに変更を伝播させることができます。CDCフィードはINSERT、UPDATE、DELETEイベント共に到着しますが、DLTのデフォルトの挙動では主キーでマッチし、イベント順を特定するフィールドで並び替えられたソースデータセットのすべてのレコードのINSERT、UPDATEイベントを適用します。より具体的には、ターゲットテーブルで主キーがマッチするすべての行を更新し、存在しないレコードは新規の行を挿入します。`DELETE`イベントを取り扱うには、SQLでは`APPLY AS DELETE WHEN`、Pythonでは`apply_as_deletes`を使用します。

この例では、顧客を一意に識別する`id`を主キーとして用い、ターゲットストリーミングテーブルで識別された顧客にCDCイベントを適用します。`operation_date`はソースデータセットでCDCイベントの論理的順序を保持しているので、順序を守らず到着したイベントを取り扱うために、SQLでは`SEQUENCE BY operation_date`、Pythonでは`sequence_by = col("operation_date")`を使用します。`SEQUENCE BY (あるいはsequence_by)`で使用するフィールドの値は同じキーに対するすべてのアップデートでユニークになっている必要があることに注意してください。多くの場合、タイムスタンプの情報をもつカラムによって並び替えられます。最後に、ターゲットストリーミングテーブルで`operation`、`operation_date`、`_rescued_data`の3つのカラムを除外するために、SQLでは`COLUMNS * EXCEPT (operation, operation_date, _rescued_data)`、Pythonでは`"except_column_list"= ["operation", "operation_date", "_rescued_data"]`を使用しています。`COLUMNS`句を指定しない場合、デフォルトではすべてのカラムがターゲットストリーミングテーブルに含まれます。

```sql:SQL
APPLY CHANGES INTO LIVE.customer_silver
FROM stream(LIVE.customer_bronze_clean_v)
  KEYS (id)
  APPLY AS DELETE WHEN operation = "DELETE"
  SEQUENCE BY operation_date 
  COLUMNS * EXCEPT (operation, operation_date, 
_rescued_data);  
```

```py:Python
dlt.apply_changes(
  target = "customer_silver", 
  source = "customer_bronze_clean_v", 
  keys = ["id"], 
  sequence_by = col("operation_date"), 
  apply_as_deletes = expr("operation = 'DELETE'"), 
  except_column_list = ["operation", "operation_date", "_rescued_data"])
```

利用できる句の完全なリストに関しては、[こちら](https://docs.databricks.com/data-engineering/delta-live-tables/delta-live-tables-cdc.html#requirements)をチェックしてください。本記事の発行時点では、`APPLY CHANGES INTO`クエリーや`apply_changes`関数のターゲットテーブルから読み込むテーブルはLIVEテーブルである必要があり、ストリーミングLIVEテーブルではあってはいけません。

このセクションのリファレンスとして[SQL](https://www.databricks.com/wp-content/uploads/notebooks/DB-129/2-retail-dlt-cdc-sql.html)ノートブックと[python](https://www.databricks.com/wp-content/uploads/notebooks/DB-129/2-Retail_DLT_CDC_Python.html)ノートブックを利用することができます。すべてのセルの準備ができたので、クラウドオブジェクトストレージからデータを取り込むパイプラインを作成しましょう。ワークスペースで新規タブや新規ウィンドウでジョブを開き、**Delta Live Tables**を選択します。

この記事のパイプラインでは、以下のDLTパイプライン設定を行なっています。

```json:JSON
{
     "clusters": [
        {
            "label": "default",
            "num_workers": 1
        }
    ],
    "development": true,
    "continuous": false,
    "edition": "advanced",
    "photon": false,
    "libraries": [
        {
            "notebook": {
"path":"/Repos/mojgan.mazouchi@databricks.com/Delta-Live-Tables/notebooks/1-CDC_DataGenerator"
            }
        },
        {
            "notebook": {
"path":"/Repos/mojgan.mazouchi@databricks.com/Delta-Live-Tables/notebooks/2-Retail_DLT_CDC_sql"
            }
        }
    ],
    "name": "CDC_blog",
    "storage": "dbfs:/home/mydir/myDB/dlt_storage",
    "configuration": {
        "source": "/tmp/demo/cdc_raw",
        "pipelines.applyChangesPreviewEnabled": "true"
    },
    "target": "my_database"
}
```

1. 新規パイプラインを作成するために**Create Pipeline**を選択します。
1. `Retail CDC Pipeline`の様な名前をつけます。
1. ここまで作成したFakerパッケージspark用いて生成したデータセットのノートブック、DLTで生成データの取り込みを行うノートブックのパスを指定します。2つ目のノートブックパスについては、お好みの言語に応じてSQLあるいはPythonのノートブックのパスを指定することができます。
1. 1つ目のノートブックで生成されたデータにアクセスするために、設定にデータセットのパスを追加します。ここでは、データを`/tmp/demo/cdc_raw/customers`に格納しているので、2つ目のノートブックで`source/customers`を参照するために、`source`を`/tmp/demo/cdc_raw/`に設定します。
1. パイプラインの結果生成されるテーブルをクエリーできる様に`Target`(オプションと指定指定できるターゲットのデータベース)を指定します。
1. DLTが生成したデータセットやパイプラインのメタデータログにアクセスできる様に、オブジェクトストレージのストレージロケーションを指定します(オプション)。
1. パイプラインのモードを[Triggered](https://qiita.com/taka_yayoi/items/6726ad1edfa92d5cd0e9#%E9%80%A3%E7%B6%9A%E3%83%91%E3%82%A4%E3%83%97%E3%83%A9%E3%82%A4%E3%83%B3%E3%83%88%E3%83%AA%E3%82%AC%E3%83%BC%E3%83%91%E3%82%A4%E3%83%97%E3%83%A9%E3%82%A4%E3%83%B3)に設定します。トリガーモードでは、DLTパイプラインはソースのすべての新規データを一度に処理し、処理が完了すると自動で計算リソースを停止します。パイプライン設定を編集する際にTriggeredとContinuousモードを切り替えることができます。JSONで`"continuous": false`を設定するのとパイプラインをトリガーモードに設定することは等価です。
1. このワークロードでは、オートパイロットオプンションのオートスケーリングを無効にすることができ、1ワーカーのみのクラスターを使用することができます。プロダクションのワークロードにおいては、オートスケーリングを有効にし、クラスターサイズに必要な最大ワーカー数を設定することをお勧めします。
1. **Start**を選択します。
1. パイプラインが作成され実行されます！

![](https://cms.databricks.com/sites/default/files/inline-images/db-129-blog-img-3.png)

**クリックすると新規タブで動画が開きます。**

https://player.vimeo.com/video/700994477?h=4e255aa1de?version=hd1080%E2%80%9D%20frameborder=

# DLTパイプラインのリネージュ観測可能性、データ品質モニタリング

すべてのDLTパイプラインのログはパイプラインのストレージロケーションに格納されます。パイプラインを作成する際にのみストレージロケーションを指定することができます。パイプラインを作成した後はストレージロケーションを変更できないことに注意してください。

[こちら](https://qiita.com/taka_yayoi/items/0cde1f2732ff859a726f#%E3%83%91%E3%82%A4%E3%83%97%E3%83%A9%E3%82%A4%E3%83%B3%E3%81%AE%E8%A6%B3%E6%B8%AC%E5%8F%AF%E8%83%BD%E6%80%A7%E3%81%8A%E3%82%88%E3%81%B3%E5%93%81%E8%B3%AA%E3%83%A2%E3%83%8B%E3%82%BF%E3%83%AA%E3%83%B3%E3%82%B0)から以前のディープダイブをチェックすることができます。この記事に関連するサンプルDLTパイプラインにおける、パイプラインの観測可能性やデータ品質モニタリングを確認するために[こちらのノートブック](https://www.databricks.com/wp-content/uploads/notebooks/DB-129/3-retail-dlt-cdc-monitoring.html)を試してみてください。

# まとめ

この記事では、Delta Live Tables(DLT)を用いることで、ユーザーがどのようにレイクハウスにチェンジデータキャプチャ(CDC)を効率的かつシームレスに実装できるのかを説明しました。DLTはパイプラインオペレーションに対する深いカケハシと、パイプラインリネージュの観察可能性、スキーマのモニタリング、パイプラインの各ステップにおける品質チェックを提供します。DLTは自動エラーハンドリングと、ストリーミングワークロードに対する最高クラスのオートスケーリング機能をサポートしており、ユーザーは自身のワークロードに必要な最適リソースを用いて高品質なデータを入手することができます。

データエンジニアは、SQLやPythonで宣言型の[APPLY CHANGES INTO API](https://qiita.com/taka_yayoi/items/5319b9b010ce4703e98a)を用いて容易にCDCを実装することができます。この新機能によって、お使いのETLパイプラインは容易に変更を特定し、これらの変更を低レーテンシーで数万のテーブルに適用することが可能となります。

:::note
**Delta Live TablesでCDCをトライする準備はできましたか？**
Delta Live Tablesがどのようにデータの変換とETLをシンプルにするのかを学ぶにはこちらのウェビナーを視聴してください。そして、ドキュメント[Delta Live Tablesによるチェンジデータキャプチャ\(CDC\)](https://qiita.com/taka_yayoi/items/5319b9b010ce4703e98a)や公式の[github](https://github.com/databricks/delta-live-tables-notebooks)をご覧ください。ご自身でパイプラインを作成するためには[こちらの動画](https://vimeo.com/700994477)のステップに従ってください！
:::

### Databricks 無料トライアル

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