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

インクリメンタル（差分処理）のコード例

このページでは、インクリメンタルな Iceberg トランスフォームのコード例を紹介します。以下で使用する Foundry API の詳細については、Iceberg を使用したインクリメンタルな Python トランスフォームと Iceberg の変更履歴と CDC パイプラインを参照してください。Iceberg テーブルをインクリメンタルに処理する概念については、Iceberg テーブルを使用したインクリメンタル処理を参照してください。

Foundry API を使用した追加専用のインクリメンタル処理

追加専用のインクリメンタルトランスフォームでは、読み取りと書き込みの両方の Foundry API が、デフォルトでインクリメンタルな動作になります。読み取りでは前回のビルド以降に追加された行が返され、書き込みでは出力を置き換えるのではなく、出力に追加されます。

ここでは、動作を明確にするために table_read_mode="append_only" オプションを含めています。追加専用がデフォルトの読み取りモードであるため、このオプションを省略しても同じ動作になります。

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


@incremental(v2_semantics=True).with_table_incremental_options(table_read_mode="append_only")
@transform.using(
    source=TableInput("/path/input"),
    output=TableOutput("/path/output")
)
def compute(source: IncrementalIcebergInput, output: IncrementalIcebergOutput):
    added_rows = source.polars()
    active = added_rows.filter(pl.col("status") == "active")
    output.write_table(active)

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


@incremental(v2_semantics=True).with_table_incremental_options(table_read_mode="append_only")
@transform.using(
    source=TableInput("/path/input"),
    output=TableOutput("/path/output")
)
def compute(ctx: LightweightContext, source: IncrementalIcebergInput, output: IncrementalIcebergOutput):
    conn = ctx.duckdb().conn
    conn.register("added_rows", source.arrow())
    query = conn.sql("SELECT * FROM added_rows WHERE status = 'active'")
    output.write_table(query.to_arrow_table())

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


@incremental(v2_semantics=True).with_table_incremental_options(table_read_mode="append_only")
@transform.using(
    source=TableInput("/path/input"),
    output=TableOutput("/path/output")
)
def compute(source: IncrementalIcebergInput, output: IncrementalIcebergOutput):
    added_rows = source.pandas()
    active = added_rows[added_rows["status"] == "active"].astype({"status": "string"})
    output.write_table(active)

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


@incremental(v2_semantics=True).with_table_incremental_options(table_read_mode="append_only")
@transform.spark.using(
    source=TableInput("/path/input"),
    output=TableOutput("/path/output")
)
def compute(source: IncrementalTableTransformInput, output: TableTransformOutput):
    added_rows = source.dataframe()
    active = added_rows.filter(added_rows.status == "active")
    output.write_dataframe(active)

ネイティブの Iceberg 書き込み API を使用した追加専用のインクリメンタル処理

これらの例では、Foundry API を使用して追加された行を読み取り、ネイティブのオープンソース Iceberg API を使用して更新内容を追加します。

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
from transforms.api import incremental, transform
from transforms.tables import (
    IncrementalIcebergInput,
    IncrementalIcebergOutput,
    TableInput,
    TableOutput,
)
from pyiceberg.expressions import EqualTo


@incremental(v2_semantics=True).with_table_incremental_options(table_read_mode="append_only")
@transform.using(
    source=TableInput("/path/input"),
    output=TableOutput("/path/output"),
)
def compute(source: IncrementalIcebergInput, output: IncrementalIcebergOutput):
    
    # Foundry API を使用して追加された行と出力を取得します
    active = source.arrow(row_filter=EqualTo("status", "active"))
    output_table = output.table()

    # 初回実行の場合を処理します
    if output_table is None:
        output.write_table(active)
        return

    # PyIceberg の append を使用してインクリメンタルに書き込みます
    with output_table.transaction() as txn:
        txn.append(active)

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
from pyspark.sql.functions import col, days
from transforms.api import IncrementalTransformContext, incremental, transform, IncrementalTableTransformInput
from transforms.tables import TableInput, TableOutput, TableTransformOutput


@incremental(v2_semantics=True).with_table_incremental_options(table_read_mode="append_only")
@transform.spark.using(
    source=TableInput("/path/input"),
    output=TableOutput("/path/output")
)
def compute(ctx: IncrementalTransformContext, source: IncrementalTableTransformInput, output: TableTransformOutput):

    # Foundry API を使用して追加された行を取得します    
    added_rows = source.dataframe()
    active = added_rows.filter(col("status") == "active")

    # Iceberg の DataFrameWriterV2 API を使用し、列の変換結果でパーティション分割して書き込みます
    writer = active.writeTo(output.identifier).partitionedBy(days(col("ts")))

    # ctx.is_incremental を使用してインクリメンタルな解決と非インクリメンタルな解決の場合を区別します
    if ctx.is_incremental:
        writer.append()
    else:
        writer.createOrReplace()

Foundry API を使用した変更履歴のインクリメンタル処理

この例では、Foundry の変更履歴 API を使用して Iceberg の変更履歴を読み取り、一意の識別子に基づいて出力テーブルに適用する方法を示します。

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
from transforms.api import incremental, transform, IncrementalTransformContext, IncrementalTableTransformInput
from transforms.tables import TableInput, TableOutput, TableTransformOutput
from pyspark.sql import functions as F, Window


@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
):
    # 前回の実行以降の変更のみを読み込みます
    changelog_df = source.changelog(["id"])

    # 識別子ごとに最新の変更のみを保持します
    window = Window.partitionBy("id").orderBy(F.col("_change_ordinal").desc())
    latest = (
        changelog_df
        .filter(F.col("_change_type") != "UPDATE_BEFORE")
        .withColumn("_row_number", F.row_number().over(window))
        .filter(F.col("_row_number") == 1)
        .drop("_row_number")
    )

    # INSERT は upsert として機能します
    reconciled = latest.withColumn(
        "_change_type",
        F.when(
            (F.col("_change_type") != "DELETE")
            & (F.col("status") == "active"),
            F.lit("INSERT"),
        ).otherwise(F.lit("DELETE")),
    )

    # 整合性を調整した変更履歴を出力テーブルにマージします
    output.apply_changelog(reconciled, ["id"])

Iceberg API を使用した変更履歴のインクリメンタル処理

Foundry の変更履歴 API は、Iceberg の create_changelog_view Spark プロシージャ ↗ をラップしています。ラッパーで公開されていないプロシージャのオプションが必要な場合は、テーブルのカタログと識別子を使用して、このプロシージャを直接呼び出せます。このプロシージャは、オプション引数でスナップショットの範囲を渡さない限り、テーブルの最初のスナップショットから読み取ります。範囲を指定するには、どのスナップショットまで処理済みかを示すマーカーを出力テーブルで手動管理します。

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
from transforms.api import IncrementalTransformContext, incremental, transform, IncrementalTableTransformInput
from transforms.tables import TableInput, TableOutput, TableTransformOutput

_WATERMARK_PROP = "watermark.source-snapshot-id"  # 最後に処理したソースのスナップショット 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 compute(ctx: IncrementalTransformContext, source: IncrementalTableTransformInput, output: TableTransformOutput):

    spark = ctx.spark_session
    source_tbl = source.identifier
    output_tbl = output.identifier

    # このビルドで読み込む範囲の終点となるスナップショット：ソースの現在のスナップショットです。
    end = spark.sql(f"""
        SELECT snapshot_id FROM {source_tbl}.snapshots
        ORDER BY committed_at DESC LIMIT 1
    """).collect()
    end = end[0]["snapshot_id"] if end else None

    # 初回のビルド、またはインクリメンタルに解決できなかったビルド（Foundry API を使用）
    if not ctx.is_incremental:
        source_df = source.dataframe()
        output.write_dataframe(source_df.filter(source_df.status == "active"))
        # 最初のインクリメンタルビルドに開始点を設けるため、ウォーターマークを設定します
        spark.sql(f"ALTER TABLE {output.identifier} SET TBLPROPERTIES ('{_WATERMARK_PROP}' = '{end}')")
        return

    # 前回最後に処理したスナップショットを、出力テーブルのプロパティから読み戻します。
    start = spark.sql(f"""
        SHOW TBLPROPERTIES {output_tbl} ('{_WATERMARK_PROP}')
    """).collect()[0]["value"]

    if start == str(end):
        # 新しく処理するものはありません。
        return

    # 範囲 (start, end] の変更履歴を作成します
    spark.sql(f"""
        CALL {source.catalog}.system.create_changelog_view(
            table => '{source_tbl}',
            changelog_view => 'source_changelog',
            identifier_columns => array('id'),
            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 id ORDER BY _change_ordinal DESC
                    ) AS _rn
                FROM source_changelog
                WHERE _change_type != 'UPDATE_BEFORE'
            )
            WHERE _rn = 1
        ) AS src
        ON src.id = tgt.id
        WHEN MATCHED AND (src._change_type = 'DELETE' OR src.status != 'active') THEN DELETE
        WHEN MATCHED AND src.status = 'active' THEN UPDATE SET *
        WHEN NOT MATCHED AND src._change_type != 'DELETE' AND src.status = 'active' THEN INSERT *
    """)

    # 最後に読み込んだスナップショットを追跡するため、ウォーターマークを更新します
    spark.sql(f"ALTER TABLE {output_tbl} SET TBLPROPERTIES ('{_WATERMARK_PROP}' = '{end}')")

←

PREVIOUSIcebergの増分トランスフォーム

NEXT変更ログの概要

→