技術入門：Python トランスフォームでの Iceberg 変更履歴

このガイドでは、Iceberg テーブルを使用する Python トランスフォームで変更履歴を利用する方法を説明します。主要な2つの変更履歴モード、その仕組み、およびよくある落とし穴を避けるためのベストプラクティスを紹介します。

変更履歴とは

変更履歴とは、2つの時点（Iceberg スナップショット ↗で表されます）の間にテーブルで発生した変更（挿入、更新、削除）のビューです。変更履歴を利用すると、前回の実行以降に変更された部分だけを処理できるため、効率的なインクリメンタル（差分処理）が可能になります。

Foundry で変更履歴を生成するには、主に2つの方法があります。

• 識別子変更履歴（推奨）：1個以上の識別子列を指定します。ほとんどのユースケースで推奨されます。
• 正味変更履歴：識別子列を指定しません。入力データに信頼できる主キーの組み合わせがない場合に使用します。

識別子変更履歴（識別子列あり）

入力データに、各行を一意に識別し、組み合わせて主キーを構成する1個以上の列がある場合は、「識別子変更履歴」モードを使用します。このモードは「正味変更履歴」モードよりも高いパフォーマンスを発揮し、更新前と更新後のレコードを含む、より豊かなセマンティクスを提供します。

識別子変更履歴の構文

識別子変更履歴を生成するには、source.changelog を実行し、主キーの配列を指定します。

Copied!
1
changelog_df = source.changelog(["id"])

識別子変更履歴の実装の詳細

Foundry の「識別子変更履歴」モードは、Iceberg の create_changelog_view ↗ プロシージャの compute_updates オプションを使用して実装されています。

compute_updates を使用すると、次のようになります。

• 行の挿入と削除は、INSERT と DELETE の変更として表されます。
• 行の更新は識別子列に基づき、UPDATE_BEFORE 行と UPDATE_AFTER 行のペアとして表されます。これらは「更新前後のイメージ」↗とも呼ばれます。

Foundry の API は、トランザクション範囲内の最新の更新を取得するために、Iceberg の create_changelog_view に追加のステップを実装しています。たとえば、あるスナップショットで行の値が 1 から 2 に更新され、その後のスナップショットで 2 から 3 に更新された場合、生成される識別子変更履歴には、2 の UPDATE_BEFORE と 3 の UPDATE_AFTER のみが表示されます。

Iceberg テーブルの仕様は、メタデータ内の識別子フィールド ↗をサポートしていますが、一意性を強制しません。そのため、Transforms の変更履歴は現在、メタデータ内の識別子フィールドに依存していません。

識別子変更履歴のスキーマ

識別子変更履歴のスキーマは次のとおりです。

識別子変更履歴の変更の種類

識別子変更履歴の変更の種類は次のとおりです。

識別子変更履歴の例

識別子変更履歴の例を示します。

正味変更履歴（識別子列なし）

入力データのエントリーを一意に識別する識別子列の組み合わせがない場合は、「正味変更履歴」モードを使用します。この方法は「識別子変更履歴」モードよりも柔軟性がありますが、その分パフォーマンスが低下します。

正味変更履歴の構文

正味変更履歴を生成するには、次のようにします。

Copied!
1
changelog_df = source.changelog()

正味変更履歴の実装の詳細

Foundry の「正味変更履歴」モードは、Iceberg の create_changelog_view ↗ プロシージャの net_changes オプションを使用して実装されています。

Iceberg では、行の更新は行の削除と挿入として実装されています。削除と挿入が意味的に同じ行を指していることを結び付けるものがないため、行の更新という概念自体は存在しません。

「正味変更履歴」は、行の更新を2行のペアとして表す、最も単純な種類の変更履歴です。一方の行の _change_type の値は DELETE、もう一方の行の _change_type の値は INSERT です。（削除は DELETE の1行で、挿入は INSERT の1行で表されます。）

「正味変更」の「正味」は、行に対する変更（挿入、削除、更新）をスナップショットの範囲全体でまとめることを指します。たとえば、あるスナップショットで行の値が 1 から 2 に更新され、その後のスナップショットで 2 から 3 に更新された場合、「正味変更履歴」には 1 の DELETE と 3 の INSERT が表示されます。2 への更新はスキップされます。

正味変更履歴のスキーマ

正味変更履歴のスキーマは次のとおりです。

正味変更履歴の変更の種類

正味変更履歴の変更の種類は次のとおりです。

正味変更履歴の例

正味変更履歴の例を示します。

変更履歴の変化の例

この例では、スナップショット V0 から V3 にかけて Iceberg テーブルがどのように変化し、変更履歴がどのように生成されるかを示します。

スナップショット間でのテーブルの変化の例

識別子変更履歴の変化の例

次のテーブルは、Iceberg テーブルの例について、スナップショット間で識別子変更履歴がどのように変化するかを示します。

識別子変更履歴の例: V0 -> V1

識別子変更履歴の例: V1 -> V2

識別子変更履歴の例: V0 -> V2

識別子変更履歴の例: V0 -> V3（失敗するケース）

この例では、主キーの値が ID1 の行が重複しているため、V0->V3 の識別子変更履歴の生成に失敗します。このことから、識別子変更履歴モードでは識別子列の一意性が重要であることがわかります。

正味変更の変更履歴の変化の例

次のテーブルは、Iceberg テーブルの例について、スナップショット間で正味変更の変更履歴がどのように変化するかを示します。

正味変更の変更履歴の例: V0 -> V1

正味変更の変更履歴の例: V1 -> V2

正味変更の変更履歴の例: V0 -> V2

正味変更の変更履歴を作成する際に、取り消し線の付いた行は削除されます。

正味変更の変更履歴の例: V0 -> V3

取り消し線の付いた行を削除した後の正味変更：

一意性の要件とよくある落とし穴

変更履歴を扱うときは、ソーステーブルと宛先テーブルの2段階で、識別子列の一意性を確保する必要があります。

ソーステーブルの一意性（source.changelog([identifier_cols])）

次の呼び出しを行う場合：

Copied!
1
changelog_df = source.changelog(["id"])

API は、識別子列（ここでは id）がソーステーブルの各行を一意に識別することを前提としています。

ソースに同じ識別子の行が重複している場合、変更履歴の作成に失敗します。

ValueError: Duplicate rows found for identifier columns ['id'] in changelog view.

ソースの一意性を保証できない場合は、代わりに正味変更の変更履歴を使用します。

Copied!
1
changelog_df = source.changelog()  # 識別子列なし

宛先テーブルの一意性（output.apply_changelog(changelog_df, keys)）

次の呼び出しを行う場合：

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

keys 引数は、宛先テーブルの識別子列を指定します。これらの列は、変更履歴のデータフレームの識別子列と一致する必要があります。宛先テーブルでも、これらの列をキーとして各行が一意に識別される必要があります。

宛先テーブルに同じキーと一致する行が複数ある場合、または変更履歴のデータフレームにそのキーの重複が含まれている場合、apply_changelog に失敗します。

Exception: apply_changelog failed: duplicate keys detected in changelog dataframe for identifier columns ['id']

または

Exception: apply_changelog failed: multiple rows in destination table match key ['id']

変更履歴を適用する前に重複を除去する

パフォーマンスとセマンティクスの利点から識別子変更履歴を使用する場合は、apply_changelog を呼び出す前にデータの重複を必ず除去する必要があります。次に例を示します。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
from pyspark.sql import functions as F
from pyspark.sql.window import Window

# 識別子ごとに行をグループ化し、変更順序で並べ替えて、最新の
# 行が先頭に来るようにします（`_row_number == 1`）。これにより、正味の変更を表す
# 変更履歴を絞り込み、識別子ごとに最新の行のみを保持できます。
window = Window.partitionBy("id").orderBy(F.col("_change_ordinal").desc())
deduped_df = (
        changelog_df
        .withColumn("_row_number", F.row_number().over(window))
        .filter("_row_number = 1")
        .drop("_row_number")
)

output.apply_changelog(deduped_df, ["id"])

この手順を省略し、重複が存在する場合は、上記のとおり apply_changelog に失敗します。

概要とベストプラクティス

• ソースの識別子列が一意である場合は、識別子変更履歴を優先して使用します。
• ソースに重複がある場合、またはソースの主キーを信頼できない場合は、正味変更の変更履歴を使用します。
• エラーを避けるため、apply_changelog の前に必ず変更履歴のデータフレームの重複を除去し、宛先テーブルでもキーによって各行が一意に識別されることを確認します。
• ソーステーブルは copy-on-write のまま保持します。変更履歴のスキャンは、merge-on-read によって生成される削除ファイルをサポートしていません。
• 各変更履歴モードのセマンティクスとパフォーマンスのトレードオフを理解します。

詳しくは、変更履歴ビューに関する Iceberg ドキュメント ↗を参照してください。