네이티브 Iceberg API 사용

일반적인 필터와 집계는 DataFrame API로 처리할 수 있습니다. 행 수준 편집, Iceberg Spark 프로시저, 스냅샷 및 속성 조회가 필요하면 런타임 핸들에서 네이티브 식별자나 Table 객체를 얻습니다.

Spark SQL의 테이블 위치에는 identifier를, PyIceberg API에는 Table 객체를 전달합니다. 네이티브 PyIceberg 경로는 CSE 및 클라이언트 버전 제약을 따릅니다. 행을 논리적으로 편집해도 실제 파일 처리는 copy-on-write 또는 merge-on-read 설정에 따라 달라집니다.

Spark 행 편집

예제 입력에는 NULL이 없는 고유 id, 날짜 비교가 가능한 as_of, 문자열 status가 있다고 가정합니다. 같은 대상 행에 여러 입력 행이 매칭되지 않도록 준비합니다. 기준일과 상태 값은 업무 규칙 예시이며 실제 요구에 맞춰 선택합니다.

from transforms.api import TransformContext, transform
from transforms.tables import TableInput, TableOutput, TableTransformInput, TableTransformOutput


@transform.spark.using(
    updates=TableInput("/Examples/iceberg/row-updates"),
    output=TableOutput("/Examples/iceberg/edited-spark"),
)
def edit_rows_spark(ctx: TransformContext, updates: TableTransformInput, output: TableTransformOutput):
    spark = ctx.spark_session
    if not spark.catalog.tableExists(output.identifier):
        output.write_dataframe(updates.dataframe())
        return

    spark.sql(f"""
        MERGE INTO {output.identifier} AS target
        USING {updates.identifier} AS source
        ON target.id = source.id
        WHEN MATCHED THEN UPDATE SET *
        WHEN NOT MATCHED THEN INSERT *
    """)
    spark.sql(f"DELETE FROM {output.identifier} WHERE as_of < '2026-04-01'")
    spark.sql(f"UPDATE {output.identifier} SET status = 'Archived' WHERE status = 'Stale'")

출력이 없는 첫 실행은 입력으로 테이블을 초기화하고 끝납니다. 이후 실행에서 merge, 날짜 삭제와 상태 변경을 적용합니다. 이 흐름은 완전한 미러링이 아니라 지정한 행 편집을 보여 줍니다.

날짜 파티션과 증분 쓰기

가상 입력 데이터셋의 created_at을 날짜 파티션으로 사용하는 예제입니다. 입력 데이터셋과 출력 테이블은 서로 다른 리소스 경로를 사용합니다.

from pyspark.sql.functions import col, days
from transforms.api import Input, IncrementalTransformContext, IncrementalTransformInput, incremental, transform
from transforms.tables import TableOutput, TableTransformOutput


@incremental(v2_semantics=True)
@transform(
    source=Input("/Examples/iceberg/date-input-dataset"),
    output=TableOutput("/Examples/iceberg/date-partitioned-table"),
)
def partition_by_day(ctx: IncrementalTransformContext, source: IncrementalTransformInput, output: TableTransformOutput):
    writer = source.dataframe().writeTo(output.identifier)
    if ctx.is_incremental:
        writer.append()
    else:
        writer.partitionedBy(days(col("created_at"))).createOrReplace()

전체 생성 및 교체에서 파티션 명세를 정하고 증분 실행은 기존 테이블에 추가합니다. 네이티브 쓰기에서는 비증분 전환과 증분 실행을 직접 구분해야 합니다.

PyIceberg 행 편집

같은 id/as_of/status 계약을 가진 입력을 사용하는 단일 노드 예제입니다. 입력 전체를 Arrow 메모리에 읽으므로 데이터 크기를 고려합니다.

import pyarrow as pa
from transforms.api import transform
from transforms.tables import IcebergInput, IcebergOutput, TableInput, TableOutput


@transform.using(
    updates=TableInput("/Examples/iceberg/row-updates"),
    output=TableOutput("/Examples/iceberg/edited-pyiceberg"),
)
def edit_rows_pyiceberg(updates: IcebergInput, output: IcebergOutput):
    updates_arrow = updates.arrow()
    table = output.table()
    if table is None:
        output.write_table(updates_arrow)
        return

    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()
    status_field = stale.schema.field("status")
    archived = stale.set_column(
        stale.schema.get_field_index("status"),
        status_field,
        pa.array(["Archived"] * stale.num_rows, type=status_field.type),
    )
    table.overwrite(archived, overwrite_filter="status = 'Stale'")

upsert는 id로 매칭하고, delete는 조건에 맞는 행을 제거합니다. overwrite는 필터에 맞는 행을 새 행으로 대체합니다. Arrow 열을 바꿀 때 기존 Field를 사용하여 자료형과 nullability 같은 스키마 정보를 유지합니다. 첫 실행의 초기화 분기는 Spark 예제와 같습니다.

두 스냅샷의 새 식별자 비교

다음은 네이티브 스냅샷 읽기와 anti join을 사용하는 예제입니다. id가 NULL 없이 고유하고 변경되지 않는 추가 전용 데이터에서 새 식별자의 행을 찾는 용도입니다. 기존 id의 값 수정, 삭제 또는 중복 개수를 계산하는 일반 CDC는 아닙니다.

from transforms.api import transform
from transforms.tables import IcebergInput, IcebergOutput, TableInput, TableOutput


@transform.using(
    source_table=TableInput("/Examples/iceberg/append-source"),
    output_table=TableOutput("/Examples/iceberg/new-identifiers"),
)
def compare_snapshot_identifiers(source_table: IcebergInput, output_table: IcebergOutput):
    current = source_table.current_snapshot_id
    previous = source_table.previous_snapshot_id
    table = source_table.table()
    current_rows = table.scan(snapshot_id=current).to_polars()
    if previous is None:
        output_table.write_table(current_rows)
        return

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

이전 스냅샷이 없다면 현재 전체를 씁니다. 비교에 필요한 이력이 만료되면 동일한 비교를 할 수 없으므로 보존 설정을 고려합니다. 삽입, 수정과 삭제를 함께 처리하려면 Changelog API를 사용합니다.