Databricks Delta Live Tables による変更データキャプチャの簡素化

このガイドでは、Delta Live Tablesパイプラインで変更データキャプチャ（Change Data Capture）を活用して、新しいレコードを特定し、データレイク内のデータセットに加えられた変更をキャプチャする方法を説明します。Delta Live Tablesパイプラインを使用すると、スケーラブルで信頼性が高く、低レイテンシーのデータパイプラインを開発できます。同時に、データレイクでの変更データキャプチャを、最小限の計算リソースでシームレスに処理し、順不同のデータを効率的に扱えます。

変更データキャプチャの背景

変更データキャプチャ（CDC）は、データベース内の増分変更（データの削除、挿入、更新）を特定してキャプチャするプロセスです。顧客、注文、製品のステータスをニアリアルタイムのデータアプリケーション用に追跡するようなものです。CDCは、新しいイベントが発生すると、データを継続的かつ増分的に処理することで、リアルタイムのデータ進化を提供します。
2025年までに80%以上の組織がマルチクラウド戦略を導入する予定であるため、複数の環境にまたがるETLパイプラインですべてのデータ変更をリアルタイムでシームレスに一元化できる、ビジネスに適したアプローチを選択することが重要です。

CDCイベントをキャプチャすることにより、DatabricksユーザーはソーステーブルをLakehouseのDelta Tableとして再マテリアライズし、その上で分析を実行できます。また、外部システムとデータを統合することも可能です。Databricks上のDelta LakeのMERGE INTOコマンドを使用すると、データレイク内のレコードを効率的に更新および削除できます。このトピックに関する以前の詳細な解説はこちらで確認できます。これは、多くのDatabricksのお客様がDelta Lakesを活用して実行し、データレイクをリアルタイムのビジネスデータで最新の状態に保っている一般的なユースケースです。

Delta LakeはデータレイクでのリアルタイムCDC同期のための完全なソリューションを提供しますが、この度、Delta Live Tablesに変更データキャプチャ機能が追加され、アーキテクチャがさらにシンプル、効率的、かつスケーラブルになりました。DLTを使用すると、ユーザーはSQLとPythonを使用してCDCデータをシームレスに取り込むことができます。

以前のDeltaテーブルを使用したCDCソリューションではMERGE INTO操作を使用していましたが、ソースデータセットの複数の行がターゲットDeltaテーブルの同じ行を更新しようとする際に一致した場合に、障害を回避するために手動でデータを並べ替える必要がありました。順不同のデータを処理するために、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ツールが利用可能です。具体的な実装は異なりますが、これらのツールは一般的に変更履歴をログにキャプチャして記録し、下流のアプリケーションがこれらのCDCログを消費します。この例では、Debezium、FivetranなどのCDCツールからクラウドオブジェクトストレージにデータがロードされます。

さまざまなCDCツールからのデータがクラウドオブジェクトストレージやApache Kafkaのようなメッセージキューに格納されます。通常、CDCはメディアンアーキテクチャと呼ばれるものへの取り込みに使用されます。メディアンアーキテクチャは、Lakehouseでデータを論理的に整理するために使用されるデータ設計パターンであり、アーキテクチャの各レイヤーを通過するにつれてデータの構造と品質を段階的に改善することを目的としています。Delta Live Tablesを使用すると、CDCフィードからの変更をLakehouseのテーブルにシームレスに適用できます。この機能をメディアンアーキテクチャと組み合わせることで、増分変更を大規模な分析ワークロードに簡単に流すことができます。CDCとメディアンアーキテクチャを一緒に使用すると、変更または追加されたデータのみを処理すればよいため、ユーザーに多くの利点があります。これにより、ユーザーはゴールドテーブルを最新のビジネスデータで費用対効果高く最新の状態に保つことができます。

注：ここでの例は、CDCのSQLバージョンとPythonバージョンの両方に適用され、操作の使用方法にも特化しています。バリエーションを評価するには、公式ドキュメントのこちらを参照してください。

前提条件

このガイドを最大限に活用するには、以下について基本的な知識が必要です。

• SQLまたはPython
• Delta Live Tables
• ETLパイプラインの開発および/またはビッグデータシステムの使用経験
• Databricksインタラクティブノートブックとクラスター
• 新しいクラスターを作成し、ジョブを実行し、外部クラウドオブジェクトストレージまたはDBFS上の場所にデータを保存する権限を持つDatabricksワークスペースへのアクセスが必要です。
• このブログで作成するパイプラインでは、データ品質制約の適用をサポートする「Advanced」製品エディションを選択する必要があります。
   

データセット

ここでは、外部データベースからのリアルなCDCデータを消費します。このパイプラインでは、Fakerライブラリを使用して、DebeziumのようなCDCツールが生成し、Databricksでの初期取り込みのためにクラウドストレージに持ち込むことができるデータセットを生成します。Auto Loaderを使用して、クラウドオブジェクトストレージからのメッセージを増分的にロードし、生のメッセージを格納するブロンズテーブルに保存します。ブロンズテーブルは、単一の信頼できるソースへの迅速なアクセスを可能にするデータ取り込みを目的としています。次に、クリーンアップされたブロンズレイヤーテーブルからAPPLY CHANGES INTOを実行して、更新をシルバーテーブルに伝播します。データがシルバーテーブルに流れるにつれて、一般的にエンタープライズがすべての主要なビジネスエンティティのビューを提供するために、より洗練され最適化された（「必要十分な」）ものになります。下の図を参照してください。

このブログでは、顧客名、メールアドレス、住所、IDの4つのフィールドと、操作（操作コード（DELETE、APPEND、UPDATE、CREATE）を格納）、operation_date（各操作アクションのレコードの発生日時を格納）の2つのフィールドを含むJSONメッセージを必要とする簡単な例に焦点を当てています。これは変更されたデータを記述します。

上記のフィールドを持つサンプルデータセットを生成するために、フェイクデータを生成するPythonパッケージであるFakerを使用します。このデータ生成セクションに関連するノートブックはこちらで見つけることができます。このノートブックでは、生成されたデータを書き込む名前とストレージの場所を指定します。DatabricksのDBFS機能を使用します。DBFSドキュメントでその仕組みについて詳しく学んでください。次に、PySparkユーザー定義関数を使用して各フィールドの合成データセットを生成し、定義されたストレージの場所にデータを書き戻します。これは、合成データセットにアクセスするための他のノートブックで参照します。

Auto Loaderを使用した生データセットの取り込み

メディアンアーキテクチャのパラダイムによると、ブロンズレイヤーは最も生のデータ品質を保持します。この段階で、クラウドストレージの場所からAutoloaderを使用して新しいデータを増分的に読み取ることができます。ここでは、生成されたデータセットへのパスをパイプライン設定の下の設定セクションに追加します。これにより、ソースパスを変数としてロードできます。したがって、パイプライン設定の下の設定は次のようになります。

次に、この設定プロパティをノートブックにロードします。

取り込むブロンズテーブルを見てみましょう。a. SQL、b. Pythonを使用

a. SQL

b. Python

上記のステートメントは、Auto Loaderを使用して、JSONファイルからStreaming Live Tableであるcustomer_bronzeを作成します。Delta Live TablesでAutoloaderを使用する場合、スキーマやチェックポイントの場所を指定する必要はありません。これらの場所はDLTパイプラインによって自動的に管理されるためです。

Auto Loaderは、SQLではcloud_files、PythonではcloudFilesと呼ばれるStructured Streamingソースを提供します。これは、クラウドストレージのパスとフォーマットをパラメータとして受け取ります。
コンピューティングコストを削減するために、非常に低いレイテンシー要件がない限り、DLTパイプラインをマイクロバッチとしてTriggeredモードで実行することをお勧めします。

期待値と高品質なデータ

高品質で多様性があり、アクセス可能なデータセットを作成するための次のステップとして、Constraintsを使用して品質チェックの期待値基準を課します。現在、制約はretain、drop、またはfailのいずれかです。詳細については、こちらを参照してください。すべての制約は、合理化された品質監視を可能にするために記録されます。

a. SQL

b. Python

APPLY CHANGES INTOステートメントを使用して変更をダウンストリームターゲットテーブルに伝播する

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

a. SQL

b. Python

ターゲットストリーミングテーブルができたので、Apply Changes Intoクエリを使用して変更をダウンストリームターゲットテーブルに伝播できます。CDCフィードにはINSERT、UPDATE、DELETEイベントが付属していますが、DLTのデフォルトの動作は、プライマリキーで一致するソースデータセットの任意のレコードからのINSERTおよびUPDATEイベントを適用することです。これは、イベントの順序を識別するフィールドによってシーケンス化されます。具体的には、ターゲットテーブルの既存の行をプライマリキーに一致するように更新するか、ターゲットストリーミングテーブルに一致するレコードが存在しない場合に新しい行を挿入します。SQLではAPPLY AS DELETE WHENを使用するか、Pythonで同等のapply_as_deletes引数を使用してDELETEイベントを処理できます。

この例では、顧客を一意に識別し、CDCイベントがターゲットストリーミングテーブルの特定された顧客レコードに適用されるようにするために、「id」をプライマリキーとして使用しました。 「operation_date」はソースデータセット内のCDCイベントの論理的な順序を保持するため、SQLの"SEQUENCE BY operation_date"、またはPythonの同等の"sequence_by = col("operation_date")"を使用して、順序が狂って到着した変更イベントを処理します。 SEQUENCE BY (またはsequence_by)で使用するフィールド値は、同じキーに対するすべての更新で一意である必要があることに注意してください。ほとんどの場合、sequence by列はタイムスタンプ情報を持つ列になります。

最後に、SQLの"COLUMNS * EXCEPT (operation, operation_date, _rescued_data)"、またはPythonの同等の"except_column_list"= ["operation", "operation_date", "_rescued_data"]を使用して、「operation」、「operation_date」、「_rescued_data」の3つの列をターゲットストリーミングテーブルから除外しました。 "COLUMNS"句を指定しない場合、デフォルトですべての列がターゲットストリーミングテーブルに含まれます。

a. SQL

b. Python

利用可能な句の完全なリストを確認するには、こちらを参照してください。
このブログの公開時点では、APPLY CHANGES INTOクエリまたはapply_changes関数のターゲットから読み取るテーブルはライブテーブルである必要があり、ストリーミングライブテーブルにすることはできないことに注意してください。

参考として、SQLおよびPythonのノートブックが利用可能です。これで、すべてのセルが準備できたので、クラウドオブジェクトストレージからデータをインポートするパイプラインを作成しましょう。ワークスペースで新しいタブまたはウィンドウでJobsを開き、「Delta Live Tables」を選択します。

このブログに関連付けられたパイプラインには、次のDLTパイプライン設定があります。

1. 「Create Pipeline」を選択して、新しいパイプラインを作成します
2. 「Retail CDC Pipeline」などの名前を指定します
3. 前に作成したノートブックのパスを指定します。1つはFakerパッケージを使用した生成データ用、もう1つはDLTでの生成データの取り込み用です。2番目のノートブックパスは、SQLで記述されたノートブック、または選択した言語に応じてPythonで記述されたノートブックを参照できます。
4. 前のノートブックで生成されたデータにアクセスするには、設定にデータセットパスを追加します。ここでは "/tmp/demo/cdc_raw/customers" にデータを保存したので、2番目のノートブックで "source/customers" を参照するために "source" を "/tmp/demo/cdc_raw/" に設定します。
5. ターゲット（オプションで、ターゲットデータベースを参照します）を指定します。ここで、パイプラインの結果のテーブルをクエリできます。
6. オブジェクトストレージ内のストレージの場所（オプション）を指定します。ここで、DLTが生成したデータセットとパイプラインのメタデータログにアクセスできます。
7. パイプラインモードをTriggeredに設定します。Triggeredモードでは、DLTパイプラインはソース内の新しいデータを一度にすべて消費し、処理が完了するとコンピューティングリソースを自動的に終了します。パイプライン設定を編集する際に、TriggeredモードとContinuousモードを切り替えることができます。JSONで "continuous": false を設定することは、パイプラインをTriggeredモードに設定することと同じです。
8. このワークロードでは、Autopilot Optionsの下で自動スケーリングを無効にし、1つのワーカークラスターのみを使用できます。本番ワークロードでは、自動スケーリングを有効にし、クラスターサイズに必要な最大ワーカー数を設定することをお勧めします。
9. "Start" を選択します
10. パイプラインが作成され、現在実行中です！

DLT パイプラインの系統、オブザーバビリティ、およびデータ品質モニタリング

すべての DLT パイプラインログは、パイプラインのストレージの場所に保存されます。ストレージの場所は、パイプラインを作成するときにのみ指定できます。パイプライン作成後は、ストレージの場所を変更できないことに注意してください。

このトピックに関する以前の深掘り調査はこちらで確認できます。このブログに関連する DLT パイプラインのオブザーバビリティとデータ品質モニタリングを確認するには、このノートブックをお試しください。

結論

このブログでは、Delta Live Tables (DLT) を使用して、ユーザーが変更データキャプチャ (CDC) を Lakehouse プラットフォームに効率的に実装できるようにした方法を示しました。DLT は、パイプライン操作の深い可視性、パイプライン系統の監視、スキーマの監視、およびパイプラインの各ステップでの品質チェックを備えた、組み込みの品質管理を提供します。DLT は、自動エラー処理と、ストリーミングワークロード向けのクラス最高の自動スケーリング機能をサポートしており、ユーザーは品質の高いデータを、ワークロードに必要な最適なリソースで利用できます。

データエンジニアは現在、SQL または Python の DLT で、新しい宣言型のAPPLY CHANGES INTO API を使用して、CDC を簡単に実装できます。この新しい機能により、ETL パイプラインは変更を簡単に特定し、低レイテンシサポートで数万のテーブルにわたってそれらの変更を適用できます。

(このブログ記事はAI翻訳ツールを使用して翻訳されています) 原文記事