注: 以下の翻訳の正確性は検証されていません。AIPを利用して英語版の原文から機械的に翻訳されたものです。

例：Python トランスフォームの変更履歴ビュー

このページでは、Python トランスフォームで CDC（Change Data Capture）と Iceberg の変更履歴を扱うためのコード例を紹介します。

特に、次のパターンとベストプラクティスに注意してください。

Iceberg テーブルを扱う際は、必ず @incremental(v2_semantics=True) を使用します。
デコレーターに .with_table_incremental_options(table_read_mode="changelog") を追加します。これがない場合、トランスフォームは入力を追加専用モードで解決し、source.changelog() はエラーを発生させます。詳細はインクリメンタル（差分処理）読み取りモードを参照してください。
source.changelog([identifier_columns]) を使用して、識別子列で定義される一意の行ごとに、前回の実行以降の変更を取得します。
マージする前に、変更履歴にロジック（クレンジング、重複排除、エンリッチメントなど）を適用します。
output.apply_changelog(changelog_view_df, [identifier_columns]) を使用して、変更（アップサートと削除を含む）をマージします。
インクリメンタルのコンテキストが利用できない場合（初回実行時など）にのみ、完全なスナップショットによる上書きに切り替えます。

例：インクリメンタル（差分）レプリケーション

この例は、必要な絞り込みを適用した後に変更履歴 API を使用して、ソースの Iceberg テーブルから対象テーブルに変更をレプリケートする、シンプルな CDC ワークフローです。

Copied!1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
from transforms.api import (
    IncrementalTableTransformInput,
    IncrementalTransformContext,
    incremental,
    transform,
)
from transforms.tables import TableInput, TableOutput, TableTransformOutput

@incremental(v2_semantics=True).with_table_incremental_options(table_read_mode="changelog")
@transform.spark.using(
    source=TableInput("<PATH>/input"),
    output=TableOutput("<PATH>/output"),
)
def incremental_cleanup(
    ctx: IncrementalTransformContext,
    source: IncrementalTableTransformInput,
    output: TableTransformOutput,
):
    # 最後に確認した Iceberg スナップショットIDから変更履歴ビューを作成します。
    changelog_view_df = source.changelog(["id"])

    # 変更履歴ビューを読み込み、対象の Iceberg テーブルにマージします。
    output.apply_changelog(
        changelog_view_df.filter(...),  # 任意の変換ロジックを適用した変更履歴ビュー
        ["id"]  # 識別子の列
    )

要点：

source.changelog([identifier_cols]) で取得されるのは、前回の実行以降に新規作成、変更、または削除された行のみです。
変更履歴のデータフレームは変更できますが、apply_changelog を使用する場合は、_change_type 列を保持する必要があります。
output.apply_changelog(df, identifier_columns) は変更を対象テーブルにマージします。各行は、_change_type の値と、出力テーブル内で識別子に基づいて一致する行に応じて処理されます。
- INSERT/UPDATE_AFTER：識別子列が一致する既存の行を更新し、一致する行がない場合は新しい行を挿入します。UPDATE_BEFORE は無視されます。
- DELETE：出力テーブル内で識別子列が一致する既存の行を削除します。

変更履歴のパフォーマンス最適化

変更履歴の読み取りと更新は、インクリメンタルな変更を扱う場合に最も高いパフォーマンスを発揮します。入力テーブル全体を変更履歴として読み取って適用すると、広い範囲のスナップショットにわたって行の変更を関連付ける必要があるため、処理が遅くなることがあります。パフォーマンスを向上させるには、ctx.is_incremental を使用してコードを分岐させ、トランスフォームがインクリメンタルに実行されていない場合（初回実行時など）は、完全なスナップショットによる上書きに切り替えます。
```
Copied!1
2
3
4
5
6
7
def changelog_with_fallback(ctx, source, output):
    if ctx.is_incremental:
        changelog_df = source.changelog(["id"])
        output.apply_changelog(changelog_df.filter(...), ["id"])
    else:
        # 完全なスナップショットでの上書き
        output.write_dataframe(source.dataframe().filter(...))
```
識別子列を指定すると、内部のパーティション分割が簡素化されるため、変更履歴ビューの作成パフォーマンスが向上します。識別子列を渡さない場合、システムは正味の変更を表す変更履歴を生成しますが、これは効率が低くなることがあります。正味の変更を表す変更履歴の扱い方については、変更履歴の技術入門を参照してください。