Delta Lake の Change Data Feed を使って CDC を簡略化する方法

Databricksでこのノートブックを試す
 
変更データキャプチャ（CDC）は、Databricksで多くのお客様が実装しているユースケースです。このトピックに関する以前の詳しい解説はこちらでご覧いただけます。通常、CDCは、データ取り込みから分析までのアーキテクチャで、メダリオンアーキテクチャと呼ばれるものに使用されます。メダリオンアーキテクチャは、ソースシステムから取り込まれた生データを、ブロンズ、シルバー、ゴールドのテーブルを通じてリファインします。CDCとメダリオンアーキテクチャは、変更または追加されたデータのみを処理すればよいため、ユーザーに多くのメリットをもたらします。さらに、アーキテクチャ内の異なるテーブルにより、データサイエンティストやBIアナリストなどの異なるペルソナが、それぞれのニーズに合った最新の正しいデータを使用できます。Delta Lakeの新しいエキサイティングな機能であるChange Data Feed（CDF）を発表できることを嬉しく思います。この機能により、このアーキテクチャの実装がよりシンプルになり、Delta LakeのMERGE操作とログバージョニングが可能になります！

Delta Lakeの利用を開始するためのステップバイステップガイドについては、O'Reillyの新しい電子書籍をいち早くプレビューで入手してください。 O'Reillyの新しい電子書籍

CDF機能が必要な理由

多くのお客様はDatabricksを使用してCDCを実行しています。これは、他のビッグデータテクノロジーと比較してDelta Lakeの方が実装が容易なためです。しかし、適切なツールがあっても、CDCの実行は依然として困難な場合があります。CDFは、コーディングをさらにシンプルにし、CDCに関する最大の課題に対処するために設計されました。それらの課題には以下が含まれます。

• 品質管理 - バージョン間の行レベルの変更を取得するのは困難です。
• 非効率性 - 現在のバージョンでは変更がファイルレベルであり、行レベルではないため、変更されていない行を考慮するのは非効率的になる可能性があります。

Change Data Feed（CDF）の実装がこれらの問題をどのように解決するかを以下に示します。

• シンプルさと利便性 - 変更を識別するための一般的で使いやすいパターンを使用し、コードをシンプル、便利、かつ理解しやすくします。
• 効率性 - バージョン間で変更された行のみを取得できるため、MERGE、UPDATE、DELETE操作の下流での消費が非常に効率的になります。

CDFはDeltaテーブルからの変更のみをキャプチャし、有効化後は前方のみを対象とします。

CDFの活用事例

一般的なユースケースである財務予測のCDFの例を見てみましょう。ブログの冒頭で参照されているノートブックは、財務データを摂取します。1株当たり利益（EPS）の推定値は、アナリストが企業の四半期ごとの1株当たり利益を予測した財務データです。生データは、多くの異なるソースから、複数の株式に対して複数のアナリストから取得される可能性があります。

CDF機能を使用すると、データはブロンズテーブル（生データ取り込み）に単純に挿入され、次にシルバーテーブルでフィルタリング、クリーニング、拡張され、最後に、シルバーテーブルの変更データに基づいてゴールドテーブルで集計値が計算されます。

これらの変換は複雑になる可能性がありますが、幸いなことに、行ベースのCDF機能はシンプルで効率的になりました。しかし、どのように使用するのでしょうか？詳しく見ていきましょう！

注：ここでの例は、CDFのSQLバージョンと、操作の特定の使用方法に焦点を当てています。バリエーションを評価するには、こちらのドキュメントを参照してください。

Delta LakeテーブルでのCDFの有効化

テーブルでCDF機能を利用するには、まずそのテーブルで機能を有効にする必要があります。以下は、テーブル作成時にブロンズテーブルでCDFを有効にする例です。テーブルの更新としてCDFを有効にすることもできます。さらに、クラスターで作成されたすべてのテーブルに対してクラスターでCDFを有効にすることもできます。これらのバリエーションについては、こちらのドキュメントを参照してください。

変更データのクエリ

変更データをクエリするには、table_changes操作を使用します。以下の例には、挿入された行と、更新された行の事前画像と事後画像を表す2つの行が含まれており、必要に応じて変更の違いを評価できます。削除された行に対して返されるdelete Change Typeもあります。

この例では、開始バージョンに基づいて変更されたレコードにアクセスしますが、必要に応じて、終了バージョン、および開始タイムスタンプと終了タイムスタンプに基づいてバージョンを制限することもできます。この例はSQLに焦点を当てていますが、Python、Scala、Java、Rでこのデータにアクセスする方法もあります。これらのバリエーションについては、こちらのドキュメントを参照してください。

MERGEステートメントでのCDF行データの使用

ゴールドテーブルへのマージのような集計MERGEステートメントは、本質的に複雑になる可能性がありますが、CDF機能により、これらのステートメントのコーディングがよりシンプルかつ効率的になります。

上記の図に示すように、CDFはどの行が変更されたかを簡単に特定できます。これは、table_changes操作を使用して、変更された、または新しいデータのみに必要な集計を実行するためです。以下に、変更されたデータを使用して、変更された日付と株式シンボルを特定する方法を示します。

以下に示すように、シルバーテーブルの変更データを使用して、ゴールドテーブルに更新または挿入する必要がある行のデータのみを集計できます。これを行うには、table_changes（'table_name'、'version'）でINNER JOINを使用します。

最終結果は、時間とともに段階的に変更できる、明確で簡潔なゴールドテーブルのバージョンです。

主なユースケース

新しいCDF機能の一般的なユースケースとメリットをいくつかご紹介します。

Silver & Gold テーブル

初期のMERGE比較に続く変更のみを処理することでDeltaのパフォーマンスを向上させ、ETL/ELT操作を高速化・簡素化します。

マテリアライズドビュー

変更があった箇所のみを更新することで、基盤となるテーブル全体を再処理することなく、BIや分析で使用できる最新の集計ビューを作成します。

変更の送信

Change Data FeedをKafkaやRDBMSなどの下流システムに送信し、データパイプラインの後続ステージで増分処理できるようにします。

監査証跡テーブル

Change Data Feedの出力をDeltaテーブルとしてキャプチャすることで、永続的なストレージと効率的なクエリ機能を提供し、削除が発生した場合や更新が行われた場合を含む、すべての変更を時間の経過とともに確認できます。

Change Data Feed の使用時期

結論

Databricksでは、不可能を可能にし、困難をシンプルにすることを目指しています。CDC、ログバージョニング、MERGEの実装は、Delta Lakeが作成されるまで、大規模に行うことは事実上不可能でした。そして今、エキサイティングなChange Data Feed (CDF) 機能により、さらにシンプルで効率的なものにしています！

Databricksでこのノートブックを試す

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