Iceberg 증분 변환 코드 예제

추가 전용 Foundry API 네 가지, 네이티브 추가 쓰기 두 가지와 변경 로그 두 가지를 구분합니다. 입력 및 출력 경로는 가상 값이며 실제 리소스로 바꿉니다. status 필터 예제는 문자열 status 열을 사용합니다. 변환 로직을 바꾸어 기존 출력이 유효하지 않으면 semantic_version을 올려 비증분 재계산을 수행합니다.

Foundry API의 추가 전용 처리

증분 실행에서는 일반 입력 읽기가 추가된 행을 반환하고 출력 API도 추가 쓰기를 수행합니다. 첫 실행 또는 비증분 전환에서는 전체 읽기 및 교체 의미를 적용합니다. append_only는 기본값이지만 코드에 명시합니다. with_table_incremental_options에는 transforms-tables0.1390.0 이상이 필요합니다.

Polars

import polars as pl
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("/Examples/iceberg/append-source"),
    output=TableOutput("/Examples/iceberg/append-polars"),
)
def append_polars(source: IncrementalIcebergInput, output: IncrementalIcebergOutput):
    output.write_table(source.polars().filter(pl.col("status") == "active"))

DuckDB

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("/Examples/iceberg/append-source"),
    output=TableOutput("/Examples/iceberg/append-duckdb"),
)
def append_duckdb(ctx: LightweightContext, source: IncrementalIcebergInput, output: IncrementalIcebergOutput):
    connection = ctx.duckdb().conn
    connection.register("added_rows", source.arrow())
    result = connection.sql("SELECT * FROM added_rows WHERE status = 'active'")
    output.write_table(result.to_arrow_table())

pandas

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("/Examples/iceberg/append-source"),
    output=TableOutput("/Examples/iceberg/append-pandas"),
)
def append_pandas(source: IncrementalIcebergInput, output: IncrementalIcebergOutput):
    added = source.pandas()
    active = added[added["status"] == "active"].astype({"status": "string"})
    output.write_table(active)

PySpark

from transforms.api import IncrementalTableTransformInput, incremental, transform
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("/Examples/iceberg/append-source"),
    output=TableOutput("/Examples/iceberg/append-spark"),
)
def append_spark(source: IncrementalTableTransformInput, output: TableTransformOutput):
    added = source.dataframe()
    output.write_dataframe(added.filter(added.status == "active"))

네이티브 PyIceberg 추가 쓰기

읽기는 Foundry API로 수행하고 네이티브 Table의 트랜잭션으로 추가합니다. 비증분 전환에서 전체 입력을 다시 append하면 중복이 생기므로 그 경우에는 Foundry 출력 API로 전체를 교체합니다. PyIceberg 네이티브 경로의 형식 및 CSE 지원을 확인합니다.

from pyiceberg.expressions import EqualTo
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("/Examples/iceberg/append-source"),
    output=TableOutput("/Examples/iceberg/append-native-python"),
)
def append_native_python(ctx, source: IncrementalIcebergInput, output: IncrementalIcebergOutput):
    active = source.arrow(row_filter=EqualTo("status", "active"))
    if not ctx.is_incremental:
        output.write_table(active)
        return
    table = output.table()
    if table is None:
        raise ValueError("증분 출력 테이블이 없습니다. 비증분 빌드로 초기화하세요")
    with table.transaction() as transaction:
        transaction.append(active)

네이티브 Spark 추가 쓰기

입력에는 파티션으로 사용할 ts 시각 열도 있어야 합니다. 파티션 명세는 전체 생성 또는 교체 때 지정하고 증분 실행에서는 기존 테이블에 추가합니다.

from pyspark.sql.functions import col, days
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="append_only")
@transform.spark.using(
    source=TableInput("/Examples/iceberg/append-source"),
    output=TableOutput("/Examples/iceberg/append-native-spark"),
)
def append_native_spark(ctx: IncrementalTransformContext, source: IncrementalTableTransformInput, output: TableTransformOutput):
    active = source.dataframe().filter(col("status") == "active")
    writer = active.writeTo(output.identifier)
    if ctx.is_incremental:
        writer.append()
    else:
        writer.partitionedBy(days(col("ts"))).createOrReplace()

Foundry Changelog로 필터 상태 유지

원천 id는 고유해야 합니다. 단순히 active 변경 행만 남기면 이전에 출력에 있던 행이 inactive 또는 NULL 상태로 바뀌었을 때 제거되지 않습니다. 삭제를 제외한 active 변화는 INSERT/upsert로, 나머지는 DELETE로 바꿉니다.

from pyspark.sql import functions as F
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("/Examples/iceberg/cdc-source"),
    output=TableOutput("/Examples/iceberg/cdc-active-foundry"),
)
def filter_foundry_changelog(ctx: IncrementalTransformContext, source: IncrementalTableTransformInput, output: TableTransformOutput):
    if not ctx.is_incremental:
        output.write_dataframe(source.dataframe().filter(F.col("status") == "active"))
        return
    changes = source.changelog(["id"])
    reconciled = changes.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"])

네이티브 Changelog로 필터 상태 유지

네이티브 프로시저는 UPDATE_BEFORE와 중간 변경을 포함하므로 제거 및 최신 액션 선택이 필요합니다. 이 예제는 NULL이 없는 고유 id와, UPDATE_BEFORE를 제거한 뒤 키별 동일 _change_ordinal에 액션이 하나뿐이라는 선행 조건을 사용합니다. 그 조건이 보장되지 않으면 row_number로 임의의 행을 고르지 말고 전용 Foundry API나 업무상 결정된 순서를 사용합니다.

원천 데이터에는 예제의 임시 열 이름인 _latest_action_row를 사용하지 않습니다.

끝 스냅샷은 현재 입력 브랜치의 refs에서 선택합니다. 스냅샷 ID의 숫자 크기나 테이블 전체의 최근 커밋을 시간순 기준으로 사용하지 않습니다. 출력의 watermark.source-snapshot-id는 마지막 처리 ID이며 NONE은 아직 입력 스냅샷이 없었던 상태를 나타냅니다. 그 경우 다음 입력은 전체 현재 상태로 초기화합니다.

from pyspark.sql import functions as F
from transforms.api import IncrementalTableTransformInput, IncrementalTransformContext, incremental, transform
from transforms.tables import TableInput, TableOutput, TableTransformOutput

_WATERMARK_PROP = "watermark.source-snapshot-id"
_NO_SNAPSHOT = "NONE"


@incremental(v2_semantics=True, semantic_version=1).with_table_incremental_options(table_read_mode="changelog")
@transform.spark.using(
    source=TableInput("/Examples/iceberg/cdc-source"),
    output=TableOutput("/Examples/iceberg/cdc-active-native"),
)
def filter_native_changelog(ctx: IncrementalTransformContext, source: IncrementalTableTransformInput, output: TableTransformOutput):
    spark = ctx.spark_session
    source_table = source.identifier
    output_table = output.identifier
    branch = "main" if source.branch == "master" else source.branch
    head = (
        spark.table(f"{source_table}.refs")
        .filter((F.col("name") == branch) & (F.col("type") == "BRANCH"))
        .select("snapshot_id")
        .first()
    )
    end = None if head is None else int(head["snapshot_id"])

    def store_watermark(snapshot_id):
        value = _NO_SNAPSHOT if snapshot_id is None else str(snapshot_id)
        spark.sql(f"ALTER TABLE {output_table} SET TBLPROPERTIES ('{_WATERMARK_PROP}' = '{value}')")

    def replace_from_current():
        current = source.dataframe(mode="current")
        current = current.filter(F.col("status") == "active")
        current.writeTo(output_table).createOrReplace()
        store_watermark(end)

    if not ctx.is_incremental or end is None:
        replace_from_current()
        return

    watermark = (
        spark.sql(f"SHOW TBLPROPERTIES {output_table}")
        .filter(F.col("key") == _WATERMARK_PROP)
        .select("value")
        .first()
    )
    if watermark is None:
        raise ValueError("출력 watermark가 없습니다. 비증분 빌드로 초기화하세요")
    start_value = watermark["value"]
    if start_value == _NO_SNAPSHOT:
        replace_from_current()
        return
    start = int(start_value)
    if start == end:
        return

    spark.sql(f"""
        CALL {source.catalog}.system.create_changelog_view(
            table => '{source_table}',
            changelog_view => 'source_changelog',
            identifier_columns => array('id'),
            options => map('start-snapshot-id', '{start}', 'end-snapshot-id', '{end}')
        )
    """)
    spark.sql(f"""
        MERGE INTO {output_table} AS target
        USING (
            SELECT * FROM (
                SELECT *, ROW_NUMBER() OVER (
                    PARTITION BY id ORDER BY _change_ordinal DESC
                ) AS _latest_action_row
                FROM source_changelog
                WHERE _change_type <> 'UPDATE_BEFORE'
            ) WHERE _latest_action_row = 1
        ) AS source
        ON source.id = target.id
        WHEN MATCHED AND (
            source._change_type = 'DELETE' OR source.status IS NULL OR source.status <> 'active'
        ) THEN DELETE
        WHEN MATCHED AND source._change_type IN ('INSERT', 'UPDATE_AFTER')
            AND source.status = 'active' THEN UPDATE SET *
        WHEN NOT MATCHED AND source._change_type IN ('INSERT', 'UPDATE_AFTER')
            AND source.status = 'active' THEN INSERT *
    """)
    store_watermark(end)

입력이 필터를 벗어났을 때 NULL 상태도 삭제 조건에 포함합니다. watermark는 실제 변경 적용 뒤에만 갱신합니다. 읽을 과거 스냅샷이 만료되었거나 범위가 유효하지 않으면 그 문제를 무시한 채 커서를 앞으로 보내지 않습니다. 처음 도입하거나 watermark가 없는 출력은 비증분으로 초기화합니다.

변경 로그의 중복과 순변경 의미는 기술 설명에 있습니다. 키 유일성만으로 모든 변경의 순서가 자동 결정되는 것은 아닙니다.