基盤となる Iceberg API の高度な使用法

ほとんどのトランスフォームは、標準のデータフレーム構文で記述できます。ただし、Iceberg の操作の中には、標準のデータフレーム API では記述できないものや、基盤となる、より特化した表現力の高い Iceberg API を直接使うことでメリットが得られるものがあります。

例：

行レベルの操作： テーブル全体を書き直すのではなく、行をその場で編集する UPDATE、DELETE、MERGE INTO ステートメント。
create_changelog_view などの Iceberg Spark プロシージャ。
Iceberg メタデータの検査： テーブルのスナップショット、スキーマ、プロパティの読み取り。

これらについては、ランタイムオブジェクトのハンドルを通じて、オープンソースの Iceberg API に直接アクセスできます。

コンピュートの種類	ネイティブハンドル	戻り値
PySpark (`TableTransformInput` / `TableTransformOutput`)	`.identifier`, `.catalog`	テーブルの完全修飾 Spark 識別子とカタログ名（文字列）
単一ノード (`IcebergInput` / `IcebergOutput`)	`.table()`	ネイティブの PyIceberg `Table` オブジェクト

いずれの場合も、オープンソース API がテーブルを必要とする箇所にハンドルを渡します。Iceberg Spark SQL ではテーブル名として（.identifier）、PyIceberg では Table オブジェクトとして（.table()）渡します。オープンソースの Iceberg ドキュメントでテーブル名またはテーブルオブジェクトが指定されている箇所では、代わりにネイティブハンドルを渡してください。

PySpark で行を編集する

このコード例では、Iceberg Spark 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
30
31
32
33
34
35
36
37
38
39
from transforms.api import transform, TransformContext
from transforms.tables import TableInput, TableOutput, TableTransformInput, TableTransformOutput


@transform.spark.using(
    updates=TableInput("/path/input"),
    output=TableOutput("/path/output"),
)
def compute(
    ctx: TransformContext,
    updates: TableTransformInput,
    output: TableTransformOutput,
):
    # Spark 識別子（テーブル名の代わりに使用します）
    output_id = output.identifier

    spark = ctx.spark_session
    if spark.catalog.tableExists(output_id):

        updates_id = updates.identifier

        # 一時ビューとして登録された入力を使用し、MERGE INTO でアップサートします
        spark.sql(f"""
            MERGE INTO {output_id} AS t
            USING {updates_id} AS s
            ON t.id = s.id
            WHEN MATCHED THEN UPDATE SET *
            WHEN NOT MATCHED THEN INSERT *
        """)

        # 述語に基づいて行を削除します
        spark.sql(f"DELETE FROM {output_id} WHERE as_of < '2026-04-01'")

        # 行をその場で更新します
        spark.sql(f"UPDATE {output_id} SET status = 'Archived' WHERE status = 'Stale'")
 
    else:
        # 初回ビルド：入力からテーブルに初期データを取り込みます
        output.write_dataframe(updates.dataframe())

PySpark でパーティション分割されたインクリメンタル（差分処理）書き込みを行う

このコード例では、Iceberg Spark の DataFrameWriterV2 API ↗ を使用してテーブルに書き込む方法を示します。この例には、Palantir のインクリメンタル処理のセマンティクスと、Iceberg のパーティション変換 ↗の使用が含まれます。

Copied!1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
from pyspark.sql.functions import days, col
from transforms.api import Input, transform, incremental, IncrementalTransformContext, IncrementalTransformInput
from transforms.tables import TableOutput, TableTransformOutput


@incremental(v2_semantics=True)
@transform(
    input=Input("/path/input"),
    output=TableOutput("/path/input"),
)
def partition_by_day(
    ctx: IncrementalTransformContext,
    input: IncrementalTransformInput,
    output: TableTransformOutput,
):
    writer = input.dataframe().writeTo(output.identifier).\
        partitionedBy(days(col("created_at")))
    if ctx.is_incremental:
        writer.append()
    else:
        writer.createOrReplace()

PyIceberg で行を編集する

このコード例では、PyIceberg API ↗ を使用して、テーブル全体を書き直さずに行をその場で編集する方法を示します。.table() でネイティブの PyIceberg Table オブジェクトにアクセスし、その行レベルのメソッドを使用します。upsert() は Iceberg Spark SQL の MERGE INTO に相当する処理、delete() は述語に基づく削除、overwrite() は行の置換に使用します。

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
from transforms.api import transform
from transforms.tables import IcebergInput, IcebergOutput, TableInput, TableOutput
import pyarrow as pa


@transform.using(
    updates=TableInput("/path/input"),
    output=TableOutput("/path/output"),
)
def compute(updates: IcebergInput, output: IcebergOutput):

    # 入力をメモリーに読み込みます
    updates_arrow = updates.arrow()

    # ネイティブの PyIceberg テーブル
    table = output.table()

    if table is None:
        output.write_table(updates_arrow)
        return

    # 結合列で照合し、upsert() でアップサートします
    table.upsert(updates_arrow, join_cols=["id"])

    # 述語に基づいて行を削除します
    table.delete(delete_filter="as_of < '2026-04-01'")

    # フィルターに一致する行を新しい行に置き換えます
    stale = table.scan(row_filter="status = 'Stale'").to_arrow()
    archived = stale.set_column(
        stale.schema.get_field_index("status"),
        "status",
        pa.array(["Archived"] * stale.num_rows, type=pa.string()),
    )
    table.overwrite(archived, overwrite_filter="status = 'Stale'")

PyIceberg でスナップショットを比較する

IcebergInput で .table() を使用して基盤となる PyIceberg テーブルを返すことで、メタデータの検査など、よりきめ細かな PyIceberg API ↗ にアクセスできます。このコード例では、PyIceberg で個々のスナップショットを検査し、追加された行を手動で特定する方法を示します。

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 transform
from transforms.tables import IcebergInput, IcebergOutput, TableInput, TableOutput


@transform.using(
    source_table=TableInput("/path/input"),
    output_table=TableOutput("/path/output"),
)
def compute(source_table: IcebergInput, output_table: IcebergOutput):

    # スナップショットIDを取得します
    current = source_table.current_snapshot_id
    previous = source_table.previous_snapshot_id  
    
    # 生の PyIceberg テーブル
    iceberg_table = source_table.table()  

    current_rows = iceberg_table.scan(snapshot_id=current).to_polars()

    if previous is None:
        output_table.write_table(current_rows)
        return

    previous_rows = iceberg_table.scan(snapshot_id=previous).to_polars()
    new_rows = current_rows.join(previous_rows, on="id", how="anti")
    output_table.write_table(new_rows)

←

PREVIOUSコード例

NEXT増分とCDC / Icebergの増分処理

→