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

Iceberg の変更履歴を使用して、PySpark トランスフォームで変更データキャプチャ（CDC）パイプラインを構築します。変更履歴を使用すると、ビルドのたびに入力全体を再処理することなく、入力テーブルの挿入、更新、削除を出力テーブルに適用できます。

CDC パイプラインを構築するには、次の手順を実行します。

changelog テーブル読み取りモードでインクリメンタル（差分処理）トランスフォームを構築します。
入力テーブルから変更履歴を生成します。
出力テーブルに変更を適用します。
非インクリメンタルビルドを処理する場合を明示的に扱います。

実行可能なコードを直接確認するには、クイックスタートを参照してください。

`changelog` テーブル読み取りモードでインクリメンタルトランスフォームを構築する

通常どおりインクリメンタルトランスフォームを構築し、テーブル読み取りモードを必ず changelog に指定します。

Copied!1
@incremental(v2_semantics=True).with_table_incremental_options(table_read_mode="changelog")

changelog テーブル読み取りモードは、可能な場合にトランスフォームをインクリメンタルに処理すること、および挿入、更新、削除をインクリメンタルに処理できることを示します。

入力テーブルから変更履歴を生成する

入力テーブルから変更履歴を生成し、前回読み取った入力データのバージョンと現在のバージョンの間で発生した挿入、更新、削除を含む DataFrame を作成します。

変更履歴は、Foundry の source.changelog() API またはオープンソースの create_changelog_view Spark プロシージャ ↗ で生成できます。

Foundry の .changelog() API はオープンソースのプロシージャをラップし、次の2つの便利な機能を提供します。

スナップショット範囲の自動検出。 Foundry はトランスフォームのコンテキストから開始と終了のスナップショットを推定します。オープンソースのプロシージャでは、範囲を自分で指定します。
識別子ベースの変更履歴における最新の変更の選択。 Foundry は UPDATE_BEFORE 行を削除し、一意の識別子ごとに最新の変更を選択します。オープンソースのプロシージャには、スナップショット間の中間的な変更も含まれます。

使いやすさを重視する場合は、Foundry API の使用を検討してください。Foundry API で公開されていないオプションを使用する場合や、識別子の検証によるパフォーマンスへの影響を避ける場合は、オープンソースのプロシージャを使用します。

Foundry API とオープンソースのプロシージャは、どちらも識別子ベースの変更履歴と正味の変更をサポートしています。

方法	識別子ベースの変更履歴	正味の変更履歴
Foundry `.changelog` API	識別子を渡します（例：`source.changelog(["primary_key"])`）。一意の識別子ごとに最新の変更を返し、`UPDATE_BEFORE` 行を削除します。	識別子を省略します：`source.changelog()`。スナップショット範囲全体の正味の挿入と削除を返します。
オープンソースの `create_changelog_view` プロシージャ ↗	`identifier_columns => array('primary_key')` を渡します。`UPDATE_BEFORE` 行と `UPDATE_AFTER` 行を含め、指定したスナップショット間のすべての変更を返します。	識別子を省略し、`net_changes => true` を設定します。スナップショット範囲全体の正味の挿入と削除を返します。

識別子ベースの変更履歴と正味の変更履歴

識別子ベースの変更履歴は、識別子として指定した列名に基づいて変更をグループ化します。正味の変更履歴は、すべてのデータ列を使用して行を照合し、スナップショット範囲全体で一致する挿入と削除を相殺し、重複行の件数を保持します。Foundry の識別子ベースの変更履歴 API では、一意の識別子が必要です。ソースに同じ識別子の重複行が含まれている場合、変更履歴の作成に失敗します。キーの一意性を保証できない場合は、正味の変更履歴を使用します。両方のモードの例については、変更履歴 API の技術入門を参照してください。

出力テーブルに変更履歴を適用する

変更履歴を生成したら、Foundry の .apply_changelog() API またはオープンソースの Spark SQL の MERGE INTO 文 ↗を使用して出力に適用します。

方法	処理内容	使用する場面
Foundry `.apply_changelog`	指定した識別子で照合し、`INSERT` 行と `UPDATE_AFTER` 行をアップサートし、`DELETE` 行を削除します。出力キーごとの操作は最大1つである必要があります。	識別子に基づく標準的なアップサートと削除。
オープンソースの Spark SQL の `MERGE INTO` 文 ↗	記述した照合条件と更新、挿入、削除の各句を適用します。	マージにおけるカスタムの照合、条件、ビジネスロジック。

Foundry の API では、変更履歴の DataFrame と、出力との照合に使用する列名を渡します。

Copied!1
output.apply_changelog(changelog_df, ["primary_key"])

オープンソースの Spark SQL では、これらの操作を明示的に記述します。たとえば、キーごとに1つの操作を含む latest_changes ビューを用意した場合は、次のようになります。

Copied!1
2
3
4
5
6
MERGE INTO output_table AS tgt
USING latest_changes AS src
ON src.primary_key = tgt.primary_key
WHEN MATCHED AND src._change_type = 'DELETE' THEN DELETE
WHEN MATCHED AND src._change_type IN ('INSERT', 'UPDATE_AFTER') THEN UPDATE SET *
WHEN NOT MATCHED AND src._change_type IN ('INSERT', 'UPDATE_AFTER') THEN INSERT *

フィルターで除外される行の変更を伝播する

変換処理で行を絞り込む場合は、フィルターに一致しなくなった行が出力から削除されるようにしてください。その行の変更履歴の行を除外するだけでは、以前の出力行は変更されずに残ります。

Foundry の .apply_changelog() では、フィルターに一致しなくなった行の変更を DELETE に変換してから適用します。オープンソースの Spark SQL では、代わりに MERGE INTO の句で絞り込みのロジックを記述できます。出力を絞り込む例を参照してください。

非インクリメンタルビルドの場合を処理する

初回のビルド、またはトランスフォームをインクリメンタルに解決できない場合は、出力を入力全体で置き換えます。ctx.is_incremental を使用して、この場合とインクリメンタルビルドを区別します。

非インクリメンタルビルドでは、入力履歴全体にわたる変更履歴の生成と適用は避けてください。広いスナップショット範囲にわたって行の変更を対応付ける処理は、現在の入力を1回読み取る処理より効率が低くなることがあります。

以下のクイックスタートの例は、どちらも非インクリメンタルビルドに output.write_dataframe(source.dataframe()) を使用します。トランスフォームにビジネスロジックが含まれる場合は、書き込み前に入力全体に適用します。

クイックスタート

この例は、ソーステーブルから対象テーブルに変更を複製する、単純な恒等トランスフォームを示します。ソースにはキーごとに1行あることを前提としています。

Copied!1
2
3
4
5
6
7
8
9
10
11
12
13
14
from transforms.api import IncrementalTransformContext, 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/input"),
    output=TableOutput("/path/output"),
)
def cdc_transform(ctx: IncrementalTransformContext, source, output):
    if ctx.is_incremental:
        changelog_df = source.changelog(["primary_key"])
        output.apply_changelog(changelog_df, ["primary_key"])
    else:
        output.write_dataframe(source.dataframe())

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
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
from transforms.api import (
    IncrementalTableTransformInput,
    IncrementalTransformContext,
    incremental,
    transform,
)
from transforms.tables import TableInput, TableOutput, TableTransformOutput

_WATERMARK_PROP = "watermark.source-snapshot-id"


@incremental(v2_semantics=True).with_table_incremental_options(table_read_mode="changelog")
@transform.spark.using(
    source=TableInput("/path/input"),
    output=TableOutput("/path/output"),
)
def cdc_transform(
    ctx: IncrementalTransformContext,
    source: IncrementalTableTransformInput,
    output: TableTransformOutput,
):
    spark = ctx.spark_session
    source_tbl = source.identifier
    output_tbl = output.identifier

    # このビルドでの処理対象となるソースのスナップショットを読み取ります。
    snapshots = spark.sql(f"""
        SELECT snapshot_id FROM {source_tbl}.snapshots
        ORDER BY committed_at DESC LIMIT 1
    """).collect()
    end = snapshots[0]["snapshot_id"] if snapshots else None

    # 非インクリメンタルビルドでは、ソース全体をコピーしてウォーターマークを初期化します。
    if not ctx.is_incremental:
        output.write_dataframe(source.dataframe())
        spark.sql(f"""
            ALTER TABLE {output_tbl}
            SET TBLPROPERTIES ('{_WATERMARK_PROP}' = '{end}')
        """)
        return

    # 前回のビルドで処理された最後のソースのスナップショットを読み取ります。
    start = spark.sql(f"""
        SHOW TBLPROPERTIES {output_tbl} ('{_WATERMARK_PROP}')
    """).collect()[0]["value"]

    if start == str(end):
        return

    # 変更履歴を作成します。開始スナップショットは含まれず、終了スナップショットは含まれます。
    spark.sql(f"""
        CALL {source.catalog}.system.create_changelog_view(
            table => '{source_tbl}',
            changelog_view => 'source_changelog',
            identifier_columns => array('primary_key'),
            options => map('start-snapshot-id', '{start}', 'end-snapshot-id', '{end}')
        )
    """)

    # 更新前のイメージを除外し、キーごとに最新のアクションを保持します。
    spark.sql(f"""
        MERGE INTO {output_tbl} AS tgt
        USING (
            SELECT * FROM (
                SELECT *,
                    ROW_NUMBER() OVER (
                        PARTITION BY primary_key ORDER BY _change_ordinal DESC
                    ) AS _rn
                FROM source_changelog
                WHERE _change_type != 'UPDATE_BEFORE'
            )
            WHERE _rn = 1
        ) AS src
        ON src.primary_key = tgt.primary_key
        WHEN MATCHED AND src._change_type = 'DELETE' THEN DELETE
        WHEN MATCHED AND src._change_type IN ('INSERT', 'UPDATE_AFTER') THEN UPDATE SET *
        WHEN NOT MATCHED AND src._change_type IN ('INSERT', 'UPDATE_AFTER') THEN INSERT *
    """)

    # 変更を適用した後、ウォーターマークを進めます。
    spark.sql(f"""
        ALTER TABLE {output_tbl}
        SET TBLPROPERTIES ('{_WATERMARK_PROP}' = '{end}')
    """)

詳細と例については、次を参照してください。

←

PREVIOUSインクリメンタル処理のコード例

NEXT変更ログのコード例

→