Delta Lake に基づくデータウェアハウス システムの構築

Delta Lake と 2.0 の機能

データレイク、データウェアハウス、データレイクハウスというコンセプトは、多くの記事や技術共有で既に紹介されています。皆様もある程度ご存じかと思いますので、ここでは繰り返しません。Databricks が提唱するデータレイクハウス（Lakehouse）の主な機能を直接見ていきましょう。

・ACID トランザクション。1 つのテーブルを複数のワークフローで読み書きでき、トランザクションによりデータの正確性が保証されます。

・スキーマ強制とデータ管理。スキーマ強制はスキーマ検証とも呼ばれます。データ書き込み時に、データのスキーマがテーブルの要件を満たすかを検証し、データの品質を確保します。同時に、テーブルに対する管理・運用保守操作も行います。

・BI サポート。ウェアハウスに格納されたデータを BI システムに直接接続し、データ分析に活用できます。

・構造化データ、半構造化データ、非構造化データへの対応。データウェアハウスは統一された一元的なストレージを提供し、あらゆるタイプのデータをサポートします。

・オープン性。Parquet や ORC などのオープンなオープンソースストレージフォーマットを基盤ストレージとして採用しています。

・複数 API のサポート。SQL に加え、データフレームや機械学習などの API にも対応し、SQL では実現できないシナリオにも対応します。

・バッチとストリームの統合。ストリーミングとオフラインデータの ETL パイプラインが簡素化され、既存の管理・運用保守コストも削減されます。

・ストレージとコンピューティングの分離。すべての企業やチームがコストに関心を持っています。ストレージとコンピューティングの分離およびオンデマンドスケーリングにより、コスト管理をより効果的に実現できます。

上記の通り、レイクハウスの機能の多くはレイクフォーマットによって支えられており、これが Delta Lake、Iceberg、Hudi が台頭した主な背景と理由です。

Delta Lake の機能の進化と開発の歴史を見ていきましょう。

図に示す通り、上半分は近年のコミュニティの開発状況、下半分は EMR における Delta Lake の進捗を示しています。いくつかの重要なポイントを紹介します。まず、Databricks は 2019 年 6 月にバージョン 0.2 を初のリリース版として公開しました。2020 年には、バージョン 0.6 と 0.7 がそれぞれ Spark 2 の最終バージョンおよび Spark 3 の初版となりました。バージョン 0.7 以降、Databricks は DML SQL 構文をサポートしました。2022 年には、バージョン 1.2 およびリリースされたばかりのバージョン 2.0 で、多くの重要な新機能が追加されました。

Alibaba Cloud EMR は早期から Delta Lake をフォローアップしており、2019 年から主要な機能を実装してきました。これには一般的な SQL のカバー、Z-Order、データスキッピング機能が含まれます。同時に、メタストアの同期問題を段階的に解決し、他のプロダクトからのシームレスなアクセスを実現しました。Time Travel も Spark 2 上で早期にサポートした機能です。現在、コミュニティでは Spark 3.3 以降でしか Time Travel をサポートしていません。EMR はさらに自動バキュームおよび自動コンパクション機能も提供しています。データウェアハウスシナリオのサポートにおいて、EMR は G-SCD 方式を提案しました。Delta Lake のタイムトラベル機能を活用し、元のテーブル構造を維持したまま SCD（Slowly Changing Dimension）Type 2 シナリオを実現しました。同時に、EMR は Delta Lake CDC もサポートし、Delta テーブルを CDC のソースとして使用して増分データウェアハウスを実現可能にしました。

Delta Lake 2.0 の注目すべき主要機能について紹介します。

・Change Data Feed

・Z-Order クラスタリング

・冪等書き込み

・Drop Column

・Dynamic Partition Overwrite

・Multi-Part Checkpoint

Change Data Feed と Z-Order については後で詳しく説明します。

ここでは Drop Column に注目します。これはバージョン 1.2 でリリースされた Rename Column と組み合わせて使用できる機能です。この種のスキーマ進化は Column Mapping の機能に依存しています。Add Column と Change Column を比較して考えてみましょう。データが Delta テーブルに書き込まれると、Delta はスキーマ情報を保存します。同様に、Parquet レイヤーも同じスキーマ情報を保持します。両者は同一で、フィールド名で識別・格納されます。この実装では、単純な列の削除は実現できますが、削除後に同じ名前の列を追加した場合はどうなるでしょうか。これには Delta スキーマと Parquet スキーマの間にマッピング関係を構築し、各フィールドをグローバル一意識別子に対応付け、Parquet にこれらの一意な識別情報を格納する必要があります。この実装により、列名の変更操作を実行する際に、マッピング設定を変更するだけで対応できます。

Dynamic Partition Overwrite は、コミュニティでまだサポートされていない構文であり、説明するまでもなくよく知られています。Multi-Part Checkpoint はメタデータ読み込みの効率を向上させる機能です。チェックポイントについては後ほど詳しく説明します。

Delta Lake カーネルの分析と主要技術

1. Delta Lake のファイルレイアウト

Delta Lake のメタデータは自己管理されており、Hive Metastore などの外部メタデータストレージに依存しません。図の紹介テキストは緑色とオレンジ色の 2 つの部分に分かれています。下半分のオレンジ色は通常のデータまたはディレクトリファイルを示しており、通常のテーブルと違いはありません。パーティション構造で管理されています。違いは上半分のメタデータ部分にあります。この部分には 3 種類のファイルがあります。1 つ目は json ファイルで、各コミット後に生成される情報を記録します。コミットのたびに新しい json ファイルが生成されます。2 つ目はチェックポイント Parquet で、それ以前のチェックポイントファイルと後続の json ファイルをマージしたもので、メタデータ解析の高速化に使用されます。3 つ目は _last_checkpoint ファイルで、最後のチェックポイントのバージョン番号を格納し、読み込むべきチェックポイントファイルを素早く特定します。最初の 2 種類のファイルがメタデータの中核であることがわかります。

次に、Delta Lake メタデータの構成を詳しく見ていきましょう。

2. Delta Lake メタデータ ― 要素

まず基本概念を紹介します。テーブルは通常、データとメタデータの 2 つの部分で構成されます。メタデータは通常 Hive Metastore に格納され、データはファイルシステムに格納されます。Delta Lake のテーブルも同様です。通常のテーブルとの違いは、メタデータが自己管理され、データと共にファイルシステムの独自のディレクトリに格納される点です。また、テーブルパスに格納されたデータファイルがすべて有効とは限らず、メタデータを通じてどのデータファイルが有効でどのファイルが無効かを識別する必要があります。Delta Lake のメタデータ操作はすべて対応する Action 操作として抽象化されており、すべてのテーブルのメタデータは Action サブクラスによって実装されます。現在のアクションを見てみましょう。

・Metadata：テーブルのスキーマ、パーティション列、テーブル設定などの情報を保存します。

・AddFile：コミットで新規追加された有効なデータファイル。

・RemoveFile：コミットで無効としてマークされたファイルを削除します。

・AddCDCFile：コミットで新規追加された CDC ファイル。

・Protocol：Delta の読み書きプロトコル。異なるバージョンの Delta の互換性を管理します。

・CommitInfo：コミット操作の統計を記録し、簡単な監査を行います。

・SetTransaction：ストリーミングシンクの情報を格納します。

3. DDL/DML の構成

Action の要素を理解したら、異なる操作に対応する Action の組み合わせを知る必要があります。図のいくつかの例で説明します。まず、テーブルに対するすべての操作で CommitInfo が生成されますが、これは実際の目的よりも監査に使用されます。

次に、具体的な操作を見ていきましょう。

・Create Table。テーブルを定義するだけなので、Metadata のみを使用してテーブルのメタデータ情報を保存します。

・CTAS（Create Table As Select）。テーブル作成と同時にデータも読み込まれるため、Metadata と AddFile の両方の Action が存在します。

・Alter Table。Drop Partition を除く他の Alter Table 操作はメタデータのみを変更するため、ここではメタデータの変更だけで十分です。

・Insert/Update/Delete/Merge。Schema Evolution を伴わない DML ではメタデータは変更されないため、Metadata は発生しません。Update を例に取ると、Delta Lake はまず Update 文の where 条件に関連する可能性のあるファイルを読み込み、データをロードし、Update 文の set 部分で変更が必要な箇所を修正し、元のファイルの変更不要な部分と共に新しいファイルに書き込みます。つまり、読み込んだファイルを古いファイル（RemoveFile）としてマークし、新しく書き込んだファイルを AddFile で識別します。

4. メタデータの読み込み

次に、Action 要素を基にテーブルスナップショットを構築する方法を見ていきましょう。

まず _last_checkpoint ファイルを探します。存在しない場合は、0 番目のコミット json ファイルから最新の json メタデータファイルを読み込みます。存在する場合は、_last_checkpoint に記録されたバージョン番号を取得し、そのバージョン番号に対応するチェックポイントファイルとそれ以降のバージョンのコミット json ファイルを見つけ、バージョン順にメタデータファイルを解析します。図の 6 つのルールを通じて最新スナップショットのメタデータを取得します。最終的に最新の Protocol、Metadata、および有効な AddFile のセットを得ます。この 3 つがあれば、テーブルのメタデータとデータファイルがわかり、完全なスナップショットが構成されます。

5. Delta Lake のトランザクション

ACID トランザクションはレイクハウスの重要な機能です。Delta Lake の ACID トランザクションは、json ファイルをファイルシステムに正常にコミットすることで、そのコミットの実行成功をマークします。つまり、複数の同時書き込みストリームの中で、最初に json ファイルの 1 バージョンをファイルシステムにコミットできたストリームが、コミットに成功したストリームとなります。ストレージについてご存じであれば、Delta Lake のトランザクションの核心は、データが置かれているファイルシステムに原子性と永続性があるかどうかに依存していることがわかります。具体的に説明します。

・ファイルが一度書き込まれると、完全に可視化されるか完全に非表示のいずれかであり、不完全なデータファイルが読み書きされることはありません。

・同時に 1 つのライターのみがファイルを作成または名前変更できます。

・ファイルが書き込まれると、後続の List 操作で必ず可視化されます。

同時実行制御プロトコルとして、Delta Lake は OCC を使用しています。プロトコルの具体的な原理についてはここでは詳しく説明しません。

競合検出について、Delta Lake は複数ストリームの同時書き込みをサポートしており、これにより競合が発生する可能性があります。例で説明します。ユーザーがバージョン番号 10 のファイルを読み込み、変更データをバージョン 11 としてコミットしようとしたところ、他のユーザーが既にバージョン 11 をコミットしていたとします。この場合、バージョン 11 がユーザーのコミットしようとした情報と競合するかを検出する必要があります。競合の検出方法は、2 つのコミット間で同じファイルセットが操作されているかどうかを判断することです。競合がなければ、ユーザーはバージョン 12 としてコミットを試みます。この過程でバージョン 12 も他のユーザーにコミットされた場合は、引き続き検出を行います。競合がある場合は、直接エラーを報告して現在の書き込み操作が失敗したと判断し、強制書き込みによるダーティデータの発生を防ぎます。

6. Z-Order

Z-Order は現在注目されている技術です。これはより古い概念で、空間インデックス曲線であり、連続で交差がなく、点を空間的により集約させることができます。その中核機能は、多次元から 1 次元へのマッピング関係を実現することです。

例で説明します。図に示す通り、X と Y の 2 つの列があり、[0, 7] の範囲です。図は Z-Order のソート方法を示しており、データを 16 個のファイルに分割します。従来のソートで、まず X をソートしてから Y を線形ソートした場合、X に近い要素が 1 つのファイルにまとめられます。たとえば X が 0 の場合、Y ∈ [0, 3] の 4 つの縦方向の要素が 1 つのファイルに格納され、同様に 16 個のファイルが生成されます。この場合、4<=Y<=5 をクエリすると、下半分の 8 個のファイルすべてをスキャンする必要があります。Z-Order でソートすると、4 個のファイルだけをスキャンすればよいことがわかります。

もう 1 つの例です。2<=X<=3 かつ 4<=Y<=5 をクエリする場合、Z-Order でソートすると 1 個のファイルだけをスキャンすればよいのに対し、従来の線形ソートでは 2 個のファイル（X ∈ [2,3]、Y ∈ [4,7]）をスキャンする必要があります。Z-Order を使用することでスキャンするデータ量が半分になり、同じコンピューティングリソースでクエリ時間が半分になり、パフォーマンスが 2 倍になります。上記の例から、線形ソートは現在ソート中のフィールドの集約効果により注目するのに対し、Z-Order は空間的な集約効果により注目していることがわかります。

7. Z-Order + データスキッピング

Z-Order はファイルレイアウトを構築するだけです。効果を発揮するにはデータスキッピングと組み合わせる必要があります。両者は機能的に干渉せず、機能的な結合はありませんが、相互に補完し合う必要があります。Z-Order による良好な集約効果のファイルレイアウトがなければ、データスキッピングだけでは良好なファイルフィルタリング効果を達成できません。同様に、Z-Order だけではデータスキッピングがなく、単純なファイルレイアウトでは読み取りの高速化に役立ちません。具体的な使用プロセスは次の通りです。書き込み時にデータの Z-Order 配置を完了し、ファイルシステムに書き込み、ファイル粒度でファイルの対応するフィールドの min/max 値を抽出し、図に示す AddFile のメタデータ stats に書き込みます。クエリ時に min/max 値を使用してフィルタリングし、クエリ条件を満たす読み込み対象ファイルを選択してからデータをフィルタリングし、ファイルとデータの読み取りを削減します。

ここで注意すべき点は、クエリパターンが変更された場合、たとえば元の Z-Order が a と b の 2 つのフィールドに基づいていたが、一定期間後メインクエリがフィールド c になった場合や、ファイルが何度も書き込まれて集約効果が低下した場合、定期的に Z-Order を再実行して集約効果を維持する必要があることです。

Delta Lake のエコシステム構築

ここまで Delta Lake の基本概念について説明しました。現在のビッグデータアーキテクチャに基づいて、単一のシステムで全体のビッグデータエコシステムを構築できないこともご理解いただけたかと思います。次に、Delta Lake の現在のエコシステムがビッグデータシステムの構築にどのように役立つかを見ていきましょう。

まずオープンソースのエコシステムを見てみましょう。ビッグデータコンポーネントは、おおまかにストレージ、コンピューティング、メタデータ管理に分類できます。メタデータ管理のデファクトスタンダードは Hive Metastore です。ストレージには主に HDFS やクラウドベンダーのオブジェクトストレージがあります。各種コンピューティングエンジンには対応するストレージインターフェイスがあります。クエリについては、各エンジンのフレームワークセマンティクスや API が異なるため、各レイクフォーマットはクエリ/計算エンジンと一対一で連携サポートする必要があります。

いくつかの代表的なエンジンで現在のオープンソースエコシステムを紹介します。

Delta Lake 自体は Databricks がオープンソース化しているため、Spark へのサポートは基盤コードの実装からパフォーマンスまで良好です。ただし、一部の SQL 機能はオープンソース版では完全には公開・サポートされていません。Alibaba Cloud EMR の Delta Lake バージョンは現在、一般的な SQL をカバーしています。

Hive、Presto、Trino については、コミュニティがクエリ機能を実装していますが、現時点では書き込み機能はサポートされていません。上記 3 つのエンジンインターフェイスの実装は Delta Standalone プロジェクトに基づいており、非 Spark コンピューティング/クエリエンジンの読み書き機能と連携するためのスタンドアロン機能を抽象化しています。

コミュニティでまだ十分にサポートされていない点をいくつか紹介します。

・Spark でテーブルを作成した後、Hive などの他のエンジンで直接クエリできず、Hive 側で手動で外部テーブルを作成してクエリする必要があります。その理由は、Hive が Delta テーブルをクエリするには InputFormat を通じて実装する必要があるが、Spark 側で作成した Delta テーブルがメタデータを Hive Metastore に同期する際に正しい情報が取得できていないためです（Parquet や ORC などの他のテーブルタイプは Spark ソースコードで HiveServer クラスがハードコードされています）。正しいメタデータ同期が実現できていないのです。これは主に、Spark がこれらのシナリオを考慮せず、より良い拡張能力を実現していなかったことが原因と考えています。同時に、Delta Lake コミュニティも同期メタデータ関連のロジックをコード実装に組み込むことを望んでいませんでした。

・Hive で Delta 外部テーブルを作成する際にパーティションフィールドを指定できません。Delta 自体がパーティションテーブルであっても、Hive エンジンからは通常のテーブルとして扱われます。なお、このような設計でもパフォーマンスに差異は生じません。Delta Standalone はクエリ条件に応じてパーティションのプルーニングを適切に実行します。

Alibaba Cloud EMR は上記 2 点についてより良いサポートを提供しています。Spark でテーブルを作成すると自動的にメタデータがメタストアに同期され、Hive、Presto、Trino を通じて追加操作なしで直接クエリできます。同時に、Hive Metastore でテーブルのパーティション特性を正しく表示し、ユーザー使用時の混乱を回避しています。

さらに、Hive などの Standalone モジュールベースのクエリエンジンで Delta テーブルをクエリすると、メタデータ読み込みの効率問題が発生する可能性があります。たとえば Hive クエリでは、Delta テーブルのメタデータ読み込みは Hive CLI のローカルで完了します。メタデータが大きい場合、大量のメモリと時間を消費します。EMR では、emr マニフェストメタデータアクセラレーション機能を実装しました。各書き込み時に最新スナップショットに関連する AddFile 情報を事前にファイルシステムに書き込み、クエリ時にメタデータ読み込みをスキップして、このシナリオでのメタデータアクセラレーション問題を解決しています。

同時に、Presto/Trino 上での Time Travel クエリおよびデータスキッピング最適化もサポートしています。

最後に、Flink の書き込みについては、Delta はバージョン 0.4 からコミュニティで Flink Sink の機能をリリースし、バージョン 0.5 で Flink Source の機能をリリースしました。

次に、Delta Lake に対する Alibaba Cloud のエコシステムサポートを紹介します。現在、DataWorks、MaxCompute、Hologres が Delta テーブルをクエリできるよう接続を実現し、Alibaba Cloud のデータレイク構築サービス DLF をメタデータとして接続・サポートし、レイクとウェアハウスのより良い統合を支援しています。同時に、DLF のレイクテーブル自動管理モジュールも接続しており、これについて後ほど紹介します。

レイクフォーマットでは、バージョンの概念とバッチストリーミングの機能が導入されており、これにより一部の履歴バージョンのデータが現在のスナップショットで無効になったり、ストリーミングシナリオで小ファイルが生成されたりします。また、先ほど述べたように Z-Order の効果も時間の経過とともに低下します。これらの問題にはレイクテーブルの管理が必要です。たとえば、履歴ファイルの定期的なクリーニング、Z-Order の再実行、ファイルマージ操作などがあります。DLF ではレイクテーブル自動管理モジュールを実装しており、テーブルのバージョン更新をリアルタイムで感知し、テーブルの状態（有効ファイルの割合、平均ファイルサイズなどの指標）をリアルタイムで分析し、ポリシーセンターの事前定義ポリシーと組み合わせて対応するアクションを実行することで、透過的にユーザーのテーブル管理を支援します。同時に、レイクテーブルのライフサイクル管理も拡張しました。古いパーティションの使用頻度が低い場合、圧縮したり低コストストレージに移動したりできます。同時に、DLF のデータプロファイリングモジュールはテーブルレベルまたはパーティションレベルでさまざまな次元のリアルタイム統計を作成し、指標ライブラリに更新して、さらなるクエリアクセラレーションやレイクテーブル管理に活用します。

Delta Lake の代表的なデータウェアハウス事例

最後に、Delta Lake の代表的なデータウェアハウス事例を見ていきましょう。

SCD（Slowly Changing Dimension）は、データウェアハウスシナリオで時間とともに緩やかに変化するディメンションデータを処理するために使用されます。変化後の新しい値の処理方法に応じて、異なる SCD タイプが定義されています。ここでは Type 2 に注目します。新しい行のレコードを追加して履歴値を保存するタイプです。従来のデータベースでは、通常まずテーブルに Start 列と End 列を追加し、現在のディメンション値の有効範囲を識別します。End 値が空の場合、現在のディメンションが最新バージョンで有効であることを意味します。ステータス列を追加して、現在のディメンション値が有効かどうかを示すこともあります。より一般的には、すべての変化に注目するのではなく、固定された業務サイクルまたは期間内の最新値にのみ注目します。たとえば、ユーザーとその所在地のディメンションテーブルを作成します。ユーザー A が北京から杭州、武漢に転居したとすると、テーブルには異なる時間のユーザー A の異なる住所が記録されます。2022 年 7 月 16 日のユーザー A の所在地、すなわち武漢での最終的な所在地を知りたいのであり、ユーザー A が朝に北京から杭州へ、昼に杭州から武漢へ移動した過程に注目するのではありません。

SCD Type 2 の従来の方式は次の通りです。リアルタイムストリームを通じて増分データを継続的に取得し、増分テーブルに書き込みます。T+1 のデータがすべて処理された後、オフラインテーブルの T パーティションとマージして、オフラインテーブルの T+1 パーティションを生成します。使用する際は、固定された粒度（日など）を指定し、オフラインテーブルに基づいてパーティションフィールドで関連データをクエリするだけです。ここでの欠点は、オフラインテーブルの T と T+1 のデータが高度に冗長であり、ストレージの無駄が生じることです。同時に、オフラインと本番のワークフローが管理・運用のコストを増加させます。

Delta Lake がどのように上記の問題を解決するかを見てみましょう。先ほど、固定期間内の最新値により注目すると述べたので、これを G-SCD ― 固定粒度に基づくスローチェンジディメンションと名付けました。Delta Lake などのレイクフォーマットには複数バージョンの概念があるため、Time Travel の機能を使用して履歴スナップショットのデータをクエリでき、クエリパフォーマンスとデータの非重複ストレージを同時に確保できます。EMR の G-SCD は上記の特性を活用して構築されています。具体的なソリューションを見てみましょう。

まず、MySQL が binlog を Kafka に同期し、次に Spark Streaming がそれを消費し、最終的にデータを Delta Lake にコミットします。

全体のプロセスは通常のストリーミング書き込みと違いがないように見えますが、重要なポイントは次の通りです。

① 最後に、データとビジネススナップショット情報を一緒にコミットします。

② Spark Streaming はバッチデータをビジネススナップショットに従って分割し、各コミットに 1 つのビジネススナップショットのデータのみが含まれるようにし、処理済みスナップショットをセーブポイントとして永続的にバージョンを保持します。

G-SCD の実装で解決すべき 2 つの中核的問題があります。

① ビジネススナップショットと Delta バージョンのマッピング。図に示す通り、各コミットは特定のビジネススナップショットに関連付けられ（Delta バージョン V7 と V8 でコミットされたデータはビジネススナップショット T のデータ）、ビジネススナップショットは Delta バージョンに沿って増加することが要求されます（T-1 から T へ、T へ、T+1 へと）。これにより、7 月 15 日のデータなどのあるビジネスのスナップショットを、Time Travel の特定バージョンにマッピングしてクエリできます。

② セーブポイントとロールバック。従来の方式では、パーティションは積極的に削除しない限り失われないのに対し、レイクテーブルには履歴バージョンの自動クリーンアップ機能があります。G-SCD 方式では、すべてのバージョンを保持する必要はありませんが、特定のバージョンを削除されずに保持できるように指定したいので、ここでセーブポイント機能が必要です。もう 1 つは、データに誤りが発生することは避けられず、ある日のデータに戻ってデータを修復するバージョンバックトラッキング機能も必要です。ここでのロールバック機能は、コミュニティのバージョン 2.0 でリリースされたリストア機能に相当します。

ストリーミングデータ処理に精通している方であれば、データのドリフト問題があることにお気づきでしょう。この現象の原因は、前のスナップショットのデータが次のスナップショットサイクルまで到着しないことです。このような場合どうすればよいでしょうか。G-SCD ではビジネススナップショットが Delta バージョン上で増分的であることが要求されると述べました。同時に、この方式では上流 Kafka のパーティションがビジネススナップショットに従って厳密に順序付けられ、同じ ID のデータは同じパーティションにのみ格納されることを要求し、これにより特定の主キーのデータは絶対に順序が狂わないようにします。次にストリーミングレベルで、各バッチが同じビジネススナップショットに属するかを判断します。同じであれば直接コミットし、同じでなければビジネススナップショットサイクルが小さいデータのみをコミットし、もう一方のデータを一時的にキャッシュします。キャッシュ機構については、初めて出現する次のスナップショットデータを一時的に格納し、まず合理的なデータドリフトによる前のスナップショットデータを処理します。一定期間後にドリフトデータが出現しなくなると判断した時点で、この部分のデータをコミットします。この分割により、Delta 側の各コミットが 1 つのビジネススナップショットのデータのみに対応することを確保できます。

次に、G-SCD の利点を見てみましょう。

・バッチとストリームの統合、管理コストの削減

・ストレージリソースの完全な節約

・Spark/Delta のクエリ最適化を最大限に活用

・他の SCD Type 2 実装のように複数の補助フィールドを追加する必要がない。同時に、従来の方式では dt をパーティションモードとして使用するため、元の SQL を再利用でき、ユーザーの移行コストはゼロ

このソリューションは Alibaba Cloud の顧客により本番環境で広く使用されています。

Change Data Capture（CDC）。最後に CDC シナリオについて説明します。これには Delta 2.0 でリリースされた非常に重要な CDF 機能が含まれます。CDC はデータの変化をキャプチャ・識別し、変更されたデータを下流にコミットしてさらに処理するためのシナリオです。CDF により、テーブルまたはデータベースが変更データを出力できるようになります。CDF の出力結果は、データに対してどのような変更（insert、update、delete）が行われたかを識別し、変更前後のデータの内容を把握できます。CDF にはバージョンデータの変更の時点とバージョン番号情報も含まれます。Delta Lake で CDF を有効にするには、delta.enableChangeDataFeed を true に設定するだけです。

CDF 以前は、MySQL から binlog 形式で増分データを ODS レイヤーに更新することしかできませんでしたが、下流の DWD への更新は非効率なフル更新方式で DWS レイヤーにデータを更新することしかできませんでした。CDF 機能を持つことで、レイクフォーマットを CDC のソースとして使用し、ODS から DWD、DWS までの全リンクの増分リアルタイムデータウェアハウスを実現できます。具体的な事例を見てみましょう。

図に示す通り、データソースと 3 つのテーブルを定義しました。user_dim はディメンションテーブル、user_city_tbl はユーザーの所在地を示し、city_population_tbl は都市の常住人口を集計します。user_city_tbl は source から更新され、user_dim テーブルと join してから書き込まれます。city_population_tbl は city フィールドを集計して生成されます。ここで両方のテーブルの CDF を有効にして、どのようなデータが生成されるかを見てみましょう。たとえば、現在上流から 2 件のデータが到着し、user1 は杭州から、user5 は武漢からです。データは Merge 文を通じて user_city_tbl にロードされ、図に示す通り、user1 は既に存在するため住所情報が更新され、user5 は新規ユーザーのためデータが挿入されます。更新操作は 2 件のデータで表現されます。1 件は pre_update で更新前の古い値を示し、もう 1 件は post_update で更新後の新しい値を示します。新規挿入データについては、挿入操作を表す 1 件のデータのみが必要で、古い値はありません。削除操作の場合、CDC の現在の値は古い値を表し、新しい値はありません。

ここの出力形式は、MySQL binlog や Debezium 形式とは異なる形式を採用していることがわかります。比較すると、CDF 実装方式は下流のデータ処理によりフレンドリで、必要なすべての情報が含まれており、過度な変換が不要です。binlog や Debezium を使用する場合、json 文字列から必要な列情報を抽出する必要があります。

user_city_tbl の変更データを使用して、下流の city_population_tbl を増分更新し、最終的に city_population_tbl テーブルで bj 市の人口が 1 人減少し、hz と wh の都市人口が 1 人増加することを実現します。ここから、CDC の出力データには update または delete データの古いレコードの詳細を含める必要があることもわかります。そうでなければ、bj 市の人口を増分で更新し、データ集計操作を正確に実装することはできません。

プロセスは続きます。city_population_tbl テーブルも CDC ソースとして使用する必要がある場合、CDF 有効化後の CDC 出力情報は右側の図に示す通りです。

最後に、Delta Lake を通じた CDC の設計と実装を見てみましょう。

Delta Lake は CDF 方式を通じて CDC を実装しています。その考え方は、必要なシナリオでのみ CDC データを永続化し、既存のデータファイルを可能な限り再利用し、読み取り側と書き込み側のバランスを十分に取ることです。

従来のデータベースの中には独自の常駐サービスを持ち、バックグラウンドで直接関連データを生成でき、書き込み効率に影響を与えないものがあります。Delta Lake はデータストレージ層のデータ組織モードとしてのみ機能し、データの読み書きの実行は依然として Flink や Spark などのコンピューティングエンジン自体に依存します。すべての追加コストは現在のコミット内で完了する必要があり、これが書き込み効率に影響を与えます。

変更データを Diff や Join などのバージョン間の計算による純粋なクエリ方式でリアルタイムに計算しない理由は、もちろんクエリパフォーマンスです。ここで、CDC が見落とされがちな点を明らかにするために 1 つのシナリオを使用します。つまり、CDC は隣接する各コミット間の変化を感知する必要があり、クエリ方向の最初と最後のコミットの変化だけではないということです。レイクフォーマットの CDC は単一のコミットに基づいています。つまり、あるデータが 1 回目のコミットで 1 から 2 に変化し、2 回目のコミットで 2 から 3 に変化した場合、2 つのコミットの CDC データは 1→2→3 となるべきで、1 から 3 に直接変化するのではありません。CDC の実際の本番シナリオの中には、この機能が必要なものがあります。

設計方式において、Delta Lake は現在のコミット情報だけでは完全なデータ変更を簡単に取得できない場合にのみ、CDC を永続化する機能を提供します。ここでの完全な CDC には、前の値と新しい値、すべての操作、タイムスタンプ、およびバージョン番号情報が含まれます。これは、CDC データを履歴スナップショットデータを読み込まずに直接読み込んでロードできることを意味します。

上記の CDF の設計特性を理解すると、一部のシナリオでは CDC を永続化する必要があるが、他のシナリオでは不要であることがわかります。CDC を永続化する必要のないシナリオ、つまりどの操作が現在のコミット情報から直接 CDC データを返せるかについて説明します。2 つの例を示します。1 つ目は Insert into です。Insert into 構文で新規追加された AddFile は他のデータに影響を与えません。コミット json ファイルのメタデータには AddFile のみがあるため、これらの AddFile ファイルのデータを直接ロードし、各レコードに insert 操作 ID を追加し、タイムスタンプとバージョン情報を追加して CDC 形式に変換して返すことができます。2 つ目の例は Drop Partition で、コミュニティではサポートされていませんが Alibaba Cloud EMR でサポートされています。あるパーティションの下のすべての有効なデータを RemoveFile として識別します。コミット json ファイルを読み込むと RemoveFile のみのファイルリストが得られます。次に RemoveFile で識別されたデータファイルをロードし、各データに delete 操作 ID とタイムスタンプおよびバージョン情報を追加します。このような操作では、CDF の実装方式は書き込みオーバーヘッドを一切追加しません。既存のデータを直接再利用してロード変換を完了し、CDC データを返します。

次に、CDC の永続化が必要な場合を見てみましょう。たとえば Update 操作では、あるデータファイルの一部のデータを更新し、非更新部分と共に新しいデータファイルに書き込む必要があります。このようなシナリオでは、更新されたデータを直接 CDC 形式のデータに変換して出力し、ファイルシステムに保存する必要があります。クエリ時に、このようなコミットに対しては、含まれる CDC ファイルを直接読み込み、CDC データをロードして返すことができます。CDC データを永続化するファイルは、先ほど詳しく説明しなかった AddCDCFile アクションを通じて記録されます。