Iceberg の変更履歴と CDC パイプライン

Foundry のトランスフォームを使用すると、Apache Iceberg の変更履歴とスナップショット機能を活用して、堅牢でスケーラブルな変更データキャプチャ（CDC）パイプラインを簡単に構築できます。トランスフォームで CDC を使用すると、前回のパイプライン実行以降に新規作成、更新、削除されたレコードを効率的に処理でき、効率的でインクリメンタルかつ低遅延なデータ移動と処理を実現できます。

Foundry は、データセットに対する追加専用のインクリメンタルトランスフォームの既存サポートに加え、transforms-tables ライブラリの一部として、Iceberg テーブルの完全な CDC 処理をサポートするようになりました。この機能は、Iceberg の変更履歴ビュー ↗を活用して、Iceberg テーブルのスナップショット間の挿入、更新、削除を取得します。

CDC 処理のメリット

Iceberg テーブルで CDC を使用すると、次のようなメリットがあります。

• 効率的なインクリメンタル（差分処理）： CDC は実行のたびにデータセット全体を再処理することを避け、パフォーマンスを向上させ、コストを削減します。
• ストリーミングおよびリアルタイムのパイプライン： CDC は新しいレコードと変更されたレコードのみを処理することで、低遅延なデータ移動を実現します。
• 監査および緩やかに変化するディメンション（SCD）： CDC を使用すると、変更前後の状態を追跡し、完全な監査証跡や SCD タイプ2の実装に活用できます。

クイックスタート：Python トランスフォームで変更履歴ビューを使用する

Palantir のトランスフォーム API を使用して、Iceberg テーブルの変更履歴を読み書きできます。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
from transforms.api import incremental, transform
from transforms.tables import TableInput, TableOutput

@incremental(v2_semantics=True).with_table_incremental_options(table_read_mode="changelog")
@transform.spark.using(
    source=TableInput("<PATH>/your_iceberg_input_table"),
    output=TableOutput("<PATH>/your_iceberg_output_table"),
)
def cdc_transform(ctx, source, output):
    # 前回の実行以降の変更のみを読み込みます
    changelog_df = source.changelog(["your_primary_key"])
    # 変更履歴にビジネスロジックを適用します
    output.apply_changelog(changelog_df, ["your_primary_key"])

より詳細なガイドと例については、次のセクションにある変更履歴のコード例と技術入門を参照してください。入力に主キーがない例のウォークスルーも含まれています。