AUTO CDCを次のレベルへ：最も困難な実世界のユースケースを解決する

Change Data Capture（CDC）は、データエンジニアがSpark上で構築する最も一般的なものの1つであり、手作業で正しく行うのが最も面倒なものの1つでもあります。以前の投稿「Stop hand-coding change data capture pipelines」では、Apache™ Spark Declarative Pipelines（SDP）のAUTO CDCが、何百行もの壊れやすいMERGEロジックをいくつかのシンプルな宣言に置き換えることで、SCD Type 1、SCD Type 2、およびSnapshot CDCをどのように自動化するかを紹介しました。

パイプラインの要件が進化するにつれて、エンジニアは、これらの標準的なCDCパターンでは解決が困難な状況に直面します。

• 順序が乱れたバイテンポラル（bitemporal）タイムラインの処理
• 既存のデータを破損することなく、部分的なレコード更新を処理する
• ストレージの保持期間を超えて存続する監査可能性の維持

本日、私たちはこれらの現実世界の課題をまさに解決するためにAUTO CDCを次のレベルへと引き上げ、これらの機能をオープンソースのApache Spark 4.2に拡張します。

バイテンポラルAUTO CDCによる2軸の履歴追跡

標準的なSCD Type 2テーブルは、現実世界でいつ事実が変更されたかを教えてくれますが、特定の時点でシステムが何を信じていたかを教えてくれるわけではありません。

SEC規則17a-4およびFINRAの記録保管規則に基づき、企業は特定の時点に存在していたレコードを再構築できなければなりません。SECの記録保管に関する一斉調査だけでも、2021年以降、100社以上の企業に対して総額20億ドル以上の制裁金が科されています。難しいのは、今日の値を保存することではありません。数ヶ月後に、報告日に参照データが何を示していたか、そしてその時点でシステムが何を信じていたかを答えることです。

標準的なSCD Type 2は、事実がいつ変更されたかという1つのタイムラインを追跡します。バイテンポラルAUTO CDCは、以下の2つのタイムラインを独立して追跡します。

• ビジネス時間（別名：イベント時間または有効時間）：現実世界でその事実が実際に真であった時間。月曜日に株式シンボルが報告対象になった、四半期末に国コードが廃止された、など。
• システム時間 （別名：トランザクション時間または処理時間）：記録システムがデータを認識した時間。月曜日の変更が水曜日までパイプラインに到達しない場合があります。

各ターゲットテーブルには、システムが管理する4つの列が追加されます。ビジネス時間用の__START_ATと__END_AT、システム時間用の__SYSTEM_START_ATと__SYSTEM_END_ATです。単一の論理的な事実が、ビジネスバージョンとシステムバージョンの組み合わせごとに1つずつ、複数の物理的な行を持つことができます。これにより、いずれの軸に沿っても特定の時点の再構築が可能になります。重要な動作保証として、イベントはどちらのタイムラインでも任意の順序で到達できます。

すでに処理されたデータよりも早いビジネス時間またはシステム時間を持つ修正データが現れた場合、エンジンは単に末尾に追加するのではなく、影響を受ける履歴を書き換えます。手書きのロジックは不要で、2つのシーケンス列を宣言するだけで、エンジンが両方のインターバルを維持します。これは、シンボルマスターのようなディメンションテーブルにも、厳格な監査可能性が必要な取引履歴やセンサー測定値のようなファクトテーブルにも同様に有効です。FINRA CAT参照データに対しては、以下のようになります。

正確なSQL句はSTORED AS SCD TYPE BITEMPORALではなくSTORED AS BITEMPORALであり、SEQUENCE BYとSYSTEM SEQUENCE BYの両方が必要であることに注意してください。たとえば、Acmeの報告対象フラグが1月1日（ビジネス時間）に変更されたものの、フィードがそれを受信したのが1月5日（システム時間）だったとします。その後、1月8日に、実際の変更は1月1日であったが異なる値であったという、過去日付の修正が届きます。バイテンポラルAUTO CDCは、両方の問いに答えることができます。

1月3日の時点では、最初のクエリは何も返しません。これは、その時点でシステムが示していた内容に対する、監査可能な正しい回答です。今日実行される2番目のクエリは、修正された真実を反映します。2つの時計、2つの回答、どちらも正しいものです。シーケンス列はソート可能な型である必要があり、NULLのシーケンス値は使用できません。この機能はサーバーレスSDPまたはPro/Advanced製品エディションで動作し、現在はベータ版であるため、パイプラインをチャンネルPREVIEWに固定してください。

タイムトラベルの先へ：VACUUM後も存続する再現可能なML

モデルが参照データや特徴量データでトレーニングされる場合、再現性とは、数ヶ月後のレビューや監査の際に、モデルが使用した正確なデータセットを再構築できることを意味します。直感的にはDelta Lakeのタイムトラベルを使いたくなりますが、これはテーブルのファイル履歴のプロパティであり、永続的な記録ではありません。VACUUMは、最近のバージョンから参照されなくなったデータファイルを永続的に削除します。デフォルトの7日間の保持期間を過ぎると、トレーニング時にログに記録されたTIMESTAMP AS OFが静かに解決できなくなる可能性があります。バイテンポラルテーブルは、その履歴をファイルバージョンとしてではなく、データとして保存します。VACUUMとOPTIMIZEはファイルを圧縮しますが、論理的な履歴には一切手を加えないため、過去のすべてのビジネスバージョンやシステムバージョンは引き続きクエリ可能な行として残ります。ここから再現性を得るには2つの方法があります。2つの時点（ビジネス時間とシステム時間）をMLflowのパラメータとしてログに記録し、トレーニングクエリをその状態に固定します。

または、テーブルが現在のビューを公開している場合は、トレーニング時に単一のシステム時点をログに記録し、後でそのタイムスタンプでのシステム時間クエリを使用して再構築します。

いずれの方法でも、再現性の契約はMLflowランにおけるいくつかのタイムスタンプであり、バイテンポラル履歴は行として保存されるため、VACUUMが基盤となるファイルをクリーンアップした後でも、その契約は維持されます。

AutoCDCの部分更新が一般提供（GA）開始

すべてのChange Data Capture（CDC）ソースが、更新時に完全な行を出力するわけではありません。代わりに、多くのソースは変更されたフィールドのみを送信し、他のすべての列をNULLとして表します。特別な処理を行わないと、これらのNULL値によってターゲットテーブル内の既存のデータが意図せず上書きされてしまう可能性があります。これまでは、お客様はこの動作を回避するためにカスタムロジックを構築する必要がありました。AutoCDCの部分更新（Partial Updates）を使用すると、これが自動的に処理されるようになります。

部分更新は、更新イベントが列のサブセットのみを変更できるようにすることで、AutoCDCを拡張します。選択された列について、入力される更新内のNULL値は、既存の値を上書きするのではなく、「更新しない」と解釈されます。

これは、NULLを出力することで変更されていない値を省略するCDCソースに特に便利です。部分更新がない場合、これらのNULLはターゲットテーブル内の既存のデータを上書きしてしまいます。

たとえば、ターゲットテーブルに以下が含まれているとします：(1, 'A', 20)

受信する更新イベントには以下が含まれます： (1, NULL, 30)

デフォルトでは、AutoCDCは行を次のように更新します： (1, NULL, 30)。

部分更新（Partial Updates）を有効にすると、name内のNULLは「既存の値を変更しない」として扱われ、結果は次のようになります： (1, 'A', 30)。

部分更新を有効にするには、AutoCDCの定義にパラメータを追加するだけです。どの列を部分更新として処理するかを指定する方法は、次の3つから選択できます：

1. NULL値を無視する列のリスト：
   IGNORE NULL UPDATES ON columnList
2. NULL値を無視しない列のリスト：
   IGNORE NULL UPDATES ON * EXCEPT (columnList)
3. 行ごとに異なる可能性のあるソース列名：
   COLUMNS TO UPDATE

完全な構文、例、および使用方法のガイダンスについては、部分更新の適用に関するドキュメントを参照してください。

オープンソースへの継続的な取り組み

Spark Declarative Pipelinesはオープンソースであるため、最も広く使用されているフロータイプもオープンソースであるべきです。私たちは、まずAUTO CDC Type 1のPython APIをApache Spark 4.2に寄贈することから始めています。

私たちは、Sparkの他の機能の進化と同様に、一度限りのコード提供ではなく、レビュー済みのプロポーザルとプルリクエストのシリーズとしてこれを寄贈しました（SPIPおよびSPARK-56249を参照）。順序が狂ったデータに対する正確性は標準で組み込まれています。小さな補助テーブルが、削除のトゥームストーン（tombstone）などの早く到着したイベントの状態を追跡し、再試行されたマイクロバッチはターゲットを破損させることなく収束します。また、ストレージフォーマットではなくSparkのストリーミングとテーブルの抽象化に基づいて構築されているため、Delta LakeとApache Icebergの両方で動作します。

オープンソースにおける今後の予定：

• 次期リリースの機能： すでにSQLインターフェース（CREATE FLOW ... AS AUTO CDC INTO）をmasterにマージしており、これは次回のApache Sparkリリースで提供される予定です。
• 高度なパイプラインセマンティクス： SCD Type 2の完全な履歴管理、ネイティブな変更ログ入力、およびNULL値がターゲットデータを上書きするのを防ぐための部分更新サポートの開発が進められています。
• 信頼性とテスト： 順序が狂ったデータや冪等な再試行に関する自動テストスイートを拡張しつつ、apply-as-truncate機能を追加しています。

はじめに

バイテンポラル（bitemporal）コンプライアンスの導入、部分更新の設定、またはApache SparkでのオープンソースAutoCDCの探索など、開始するには以下のリソースを確認してください：

• バイテンポラルAUTO CDC：SDPで2軸の履歴追跡を設定する方法を学びます
• 部分更新ガイド：カスタムコードなしで欠落している更新フィールドを処理するための完全な構文と例を確認します。
• オープンソースAutoCDCプログラミングガイド：Apache Spark 4.2で利用可能な宣言型CDC機能を探索します

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