変更データキャプチャパイプラインの手コーディングを停止する

PythonでスナップショットからのAutoCDCを試したところ、以前は1,500行のコードでやっていたことをわずか4行のコードで置き換えられたことに驚きました。 — Fortune 500の航空宇宙・防衛企業 シニアデータエンジニア

変更データキャプチャ（CDC）とスローチェンジディメンション（SCD）は、モダンな分析およびAIワークロードの基盤です。チームは、ビジネスの現在のビューを維持する場合でも、完全な履歴コンテキストを保持する場合でも、運用データが変更されたときにダウンストリームテーブルを正確に保つために、これらに依存しています。

しかし実際には、CDCパイプラインは構築および運用が最も困難なパイプラインの1つであることがよくあります。チームは、更新、削除、および遅延データに対応するために、カスタムで複雑なMERGEロジックを作成し、ステージングテーブル、ウィンドウ関数、および推論が困難で、パイプラインが進化するにつれてさらに保守が難しくなるシーケンスの仮定を重ねています。

この記事では、データエンジニアやSQLの実務家が日々直面するCDCおよびSCDパターン、これらのパターンを手動で実装するのがなぜ難しいのか、そしてLakeflow Spark Declarative PipelinesのAutoCDCがそれらを宣言的に自動化する方法について説明します。さらに、価格とパフォーマンスにおける意味のある改善も提供します。

データエンジニアにとってCDCとSCDは依然として難しい

これらのパターンをよく理解しているチームでさえ、それを正確に実装し、時間の経過とともに正確さを維持することが、問題が発生する場所です。データ量が増加し、ユースケースが拡大するにつれて、パイプラインは壊れやすくなり、正確性の問題が遅れて表面化し、ダウンストリームテーブルを破損しないようにするには、小さな変更でさえ慎重な書き換えが必要になります。

SCDタイプ1テーブルの維持

SCDタイプ1テーブルは、最新の状態を反映するように既存の行を上書きします。この「単純な」ケースでさえ、すぐに課題に直面します。

• 更新が順不同で到着する
• 重複イベントは一貫して重複排除する必要がある
• 削除は正しく適用する必要がある
• ロジックは、リトライおよび再処理全体で冪等性を維持する必要がある

単純なMERGE INTOから始まったものが、ステージングテーブル、ウィンドウ関数、および推論が困難（または安全に変更が困難）なシーケンスの仮定を伴う深くネストされたロジックに進化することがよくあります。時間の経過とともに、チームはこれらのパイプラインに触れることをためらうようになります。

SCDタイプ2履歴の維持

SCDタイプ2は、さらなる複雑さを導入します。

• 行のバージョンと有効期間を追跡する
• 履歴を破損することなく遅延データ更新を処理する
• 常に正確に1つの「現在の」バージョンが存在することを保証する

ここでの間違いは、必ずしも大きな問題として現れるわけではありません。数週間後に微妙なメトリクスのずれとして現れたり、履歴テーブル全体を再構築する必要が生じたりすることがよくあります。

さまざまなソースからの変更データの抽出

すべてのシステムがクリーンなCDCログを生成するわけではありません。一部のシステムはネイティブの変更データフィードを生成しますが、そうでないシステムもあります。これは、多くの場合、データを消費するチームがアップストリームデータベースを制御していないため、チームはソーステーブルの連続するスナップショットを比較することによって変更を再構築することを余儀なくされます。

両方をサポートすることは、通常、個別の取り込みおよび処理ロジック、異なる正確性の仮定、および保守およびデバッグするコードパスが増えることを意味します。

CDCパイプラインの長期的な運用

CDCパイプラインが正確になったとしても、再処理とバックフィル、スキーマ進化、障害と再起動を乗り越える必要があります。手動で作成されたCDCロジックは、これらの現実が蓄積するにつれて時間の経過とともにますます壊れやすくなり、運用リスクと保守コストが増加します。

宣言的なデータエンジニアリングによる複雑なCDCパターンの自動化

AutoCDCは、これらの一般的なCDCおよびSCDパターンを宣言的な抽象化の背後で標準化するように設計されました。変更がどのように適用されるかをコードで記述する代わりに、チームは望むセマンティクスを宣言し、プラットフォームが順序付け、状態、および増分処理を管理します。

これにより、チームはパターンを毎回再発明するのではなく（これは一般的に宣言的プログラミング、特にSpark Declarative Pipelinesのコアバリューです）、パイプライン全体でCDCおよびSCDを実装するための、一貫性があり再現可能な方法を得ることができます。

変更データフィード（CDF）から変更レコードを処理する場合、AutoCDCは順序外のレコードを自動的に処理し、宣言されたシーケンス列に基づいて更新を正しく適用します。これが実際にはどのように機能するかを示すために、以下のサンプルCDCフィードを検討してみましょう。

最新データのみを保持するにはSCDタイプ1を選択するか、履歴データを保持するにはSCDタイプ2を選択することを忘れないでください。まずタイプ1から始めましょう。

SCDタイプ1メンテナンスの自動化（変更データフィードソース）

この例では、変更データフィードにユーザーテーブルの挿入、更新、および削除が含まれています。目標は、新しい更新が古い値を上書きする各レコードの現在のビューを維持することです。

SCDタイプ1の出力テーブル

ユーザー123（Isabel）は削除されたため、出力に表示されません。ユーザー125（Mercedes）は、SCDタイプ1が以前の値を上書きするため、最新の都市（Guadalajara）のみを表示します。

従来のこのアプローチでは、イベントの重複排除、順序付けの強制、削除の適用、およびリトライまたは遅延データ全体でパイプラインが正確であることを保証するために、カスタムMERGEロジックが必要です。AutoCDCは、この壊れやすいロジックを、シーケンス、重複排除、遅延データ、および増分処理を自動的に処理する宣言的なパイプライン定義に置き換えます。これにより、数十行のカスタムマージロジックが不要になります。

完全なコード例については、付録を参照してください。

SCDタイプ2履歴の自動化（変更データフィードソース）

多くの分析システムでは、最新の状態を保持するだけでは不十分です。チームはレコードが時間の経過とともにどのように変化したかの完全な履歴を必要とします。これはSCDタイプ2パターンであり、レコードの各バージョンは、アクティブだった期間を示す有効期間とともに保存されます。

SCDタイプ2の出力テーブル：

テーブルは完全な履歴を保持します。ユーザー123には2つのバージョンがあります（削除時にシーケンス6で終了）。ユーザー125には都市の変更を示す3つのバージョンがあります。 __END_AT = NULL のレコードは現在アクティブです。

これを手動で実装するには、以前のレコードをクローズし、新しいバージョンを挿入し、一度に1つのバージョンのみがアクティブであることを保証するために、複数ステップの MERGE ロジックが必要です。AutoCDCは、履歴列とバージョニングロジックを自動的に管理しながら、これらの遷移を宣言的に自動化し、更新が順不同で到着した場合でも正確性を保証します。

完全なコード例は 付録で確認してください

スナップショットソースからのCDCの推論

すべてのソースシステムが変更ログをエクスポートするわけではありません。多くの場合、チームはソーステーブルの定期的なスナップショットを受け取り、実行間で何が変更されたかを推測する必要があります。

従来、これには、MERGEロジックで変更を適用する前に、挿入、更新、削除を検出するためにスナップショットを手動で比較する必要があります。AutoCDCは、スナップショットベースのCDCをファーストクラスのパターンとして扱い、カスタム差分ロジックや状態管理を必要とせずに、スナップショット間の行レベルの変更を自動的に検出し、インクリメンタルに適用します。

これを手動で実装するには、スナップショット間の行レベルの変更を検出し、以前アクティブだったレコードをクローズし、更新された有効期間を持つ新しいバージョンを挿入する必要があります。AutoCDCはこれらの変更を自動的に導き出し、SCDタイプ2のセマンティクスを適用し、複数ステップのmergeロジックやカスタムスナップショット状態追跡を必要とせずにバージョン履歴を維持します。

順序付け、状態、および再処理の管理

Lakeflow Spark Declarative Pipelinesは、インクリメンタルな進行状況を自動的に追跡し、順序外のデータを処理します。パイプラインは障害から回復し、履歴データを再処理し、変更を二重に適用したり失ったりすることなく進化できます。

実際には、これにより、チームはシーケンスロジック、ウォーターマークの記録、または再処理の安全性を自分で管理する必要がなくなり、プラットフォームがそれを処理します。

新機能：大幅な価格とパフォーマンスの向上

パイプラインロジックの単純化を超えて、最近のDatabricks Runtimeの改善により、2025年11月以降、AutoCDCワークロードのパフォーマンスとコスト効率の両方で大幅な向上が実現しました。

• SCDタイプ1
  • レイテンシの約22％改善
  • コストの約40％削減
  • 約71％の純価格パフォーマンスメリット
• SCDタイプ2
  • レイテンシの約45％削減
  • インクリメンタル更新のコストの約35％削減
  • 約96％の純価格パフォーマンスメリット

これらのメリットは、大規模で継続的に実行される実際のパイプラインにとって重要です。MERGE INTOは引き続き基本的なSparkプリミティブですが、AutoCDCはそれを基盤として、データ量が増加するにつれて、順序外のデータとインクリメンタル処理をより効率的に処理します。

AutoCDCによる顧客の成功

本番環境でCDCおよびSCDパイプラインを実行しているチームは、AutoCDCが大幅な価値を提供していると明示的に述べています。

Navy Federal Credit Unionは、Lakeflow Spark Declarative PipelinesでAutoCDCを使用して、大規模なリアルタイムイベント処理を強化しています。これは、カスタムCDCコードと継続的なパイプラインメンテナンスを排除しながら、数十億のアプリケーションイベントを継続的に処理します。

Spark Declarative Pipelinesのプログラミングモデルのシンプルさとそのサービス機能の組み合わせにより、非常に迅速なターンアラウンドタイムが実現しました。 — Jian (Miracle) Zhou, Senior Engineering Manager, Navy Federal Credit Union

Blockは、Delta Lake上の変更データキャプチャとリアルタイムストリーミングパイプラインを簡素化するために、Lakeflow Spark Declarative PipelinesでAutoCDCを使用しています。手動でコーディングされたCDCおよびmergeロジックを、実装が迅速で運用が容易な宣言型アプローチに置き換えています。

Spark Declarative Pipelinesを採用したことで、ストリーミングパイプラインの定義と開発に必要な時間が数日から数時間に短縮されました。 — Yue Zhang, Staff Software Engineer, Data Foundations, Block

スイスを拠点とする大手「フードビネンス」プロバイダーであるValora Groupは、マスターデータとリアルタイム小売分析の変更データキャプチャを合理化するために、Lakeflow Spark Declarative PipelinesでAutoCDCを使用しています。カスタムCDCコードを、実装、繰り返し、およびチーム全体でのスケーリングが容易な宣言型アプローチに置き換えています。

SDPでCDCを行ったことで、コードを書く必要がなく、すべてバックグラウンドで抽象化されているため、多くのメリットがありました。AutoCDCは行数を最小限に抑えます…非常に簡単です。 — Alexane Rose, Data and AI Architect, Valora Holding

はじめに

AutoCDCは、Databricks上のLakeflow Spark Declarative Pipelinesの一部として利用できます。

詳細については、以下をご覧ください。

• AutoCDCドキュメント（SQLおよびPython）を確認してください
• SCDタイプ1、SCDタイプ2、およびスナップショットベースのCDCの例を探る

独自のパイプラインでAutoCDCを試して、手動で作成したCDCロジックを排除しましょう！

付録

SCDタイプ1の例

from delta.tables import DeltaTable
from pyspark.sql.functions import max_by, struct

# Deduplicate: keep latest record per userId
updates = (spark.read.table("cdc_data.users")
    .groupBy("userId")
    .agg(max_by(struct("*"), "sequenceNum").alias("row"))
    .select("row.*"))

# Apply SCD Type 1: upsert updates, delete deletions
(DeltaTable.forName(spark, "target")
    .alias("t")
    .merge(updates.alias("s"), "s.userId = t.userId")

    .whenMatchedDelete(condition="s.operation = 'DELETE'")
    .whenMatchedUpdate(
        condition="s.sequenceNum > t.sequenceNum",
        set={"name": "s.name", "city": "s.city", "sequenceNum": "s.sequenceNum"}
    )
    .whenNotMatchedInsertAll(condition="s.operation != 'DELETE'")
    .execute())

from pyspark import pipelines as dp
from pyspark.sql.functions import col, expr

@dp.view
def users():
    return spark.readStream.table("cdc_data.users")

dp.create_streaming_table("target")

dp.create_auto_cdc_flow(
    target="target",
    source="users",
    keys=["userId"],
    sequence_by=col("sequenceNum"),
    apply_as_deletes=expr("operation = 'DELETE'"),
    stored_as_scd_type=1
)

SCDタイプ2の例

from delta.tables import DeltaTable
from pyspark.sql.functions import col, lit, max_by, struct

# Deduplicate: keep latest record per userId
updates = (spark.read.table("cdc_data.users")
    .groupBy("userId")
    .agg(max_by(struct("*"), "sequenceNum").alias("row"))
    .select("row.*"))

# Step 1: close out active rows for records being updated or deleted
(DeltaTable.forName(spark, "target")
    .alias("t")
    .merge(
        updates.alias("s"),
        "s.userId = t.userId AND t.__END_AT IS NULL AND s.sequenceNum > t.__START_AT"
    )
    
    .whenMatchedUpdate(set={"__END_AT": "s.sequenceNum"})
    .execute())

# Step 2: insert new rows for inserts and updates (not deletes)
new_rows = (updates
    .filter("operation != 'DELETE'")
    .withColumn("__START_AT", col("sequenceNum"))
    .withColumn("__END_AT", lit(None).cast("long"))
    .drop("operation"))

new_rows.write.mode("append").saveAsTable("target")
                    

dp.create_auto_cdc_flow(
    target="target",
    source="users",
    keys=["userId"],
    sequence_by=col("sequenceNum"),
    apply_as_deletes=expr("operation = 'DELETE'"),
    stored_as_scd_type=2
)

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