Iceberg 변경 로그로 CDC 파이프라인 만들기

Iceberg 변경 로그는 두 스냅샷 사이의 삽입, 수정과 삭제를 읽어 출력 상태에 적용하는 방법입니다. PySpark 변환에서 changelog 읽기 모드를 사용하고, 입력의 변경 로그를 만든 뒤 출력에 적용합니다. 첫 실행이나 비증분 전환은 전체 현재 입력으로 처리합니다.

증분 선언

@incremental(v2_semantics=True).with_table_incremental_options(table_read_mode="changelog")를 사용합니다. with_table_incremental_options는 transforms-tables0.1390.0 이상이 필요합니다. Changelog 모드는 Spark에서만 지원하며 원천은 copy-on-write로 유지해야 합니다. 실제 delete file이 있는 구간은 현재 changelog scan에서 읽을 수 없습니다.

로그 생성 경로

Foundry API는 시작과 끝 스냅샷을 변환 문맥에서 자동으로 찾습니다. 네이티브 프로시저는 읽을 범위를 직접 지정합니다. 식별자 검증 비용이나 추가 프로시저 옵션 때문에 네이티브 경로를 선택할 수 있지만, 그 경우 키와 변경 순서의 선행 조건도 직접 충족해야 합니다.

식별자 열은 원천 행을 고유하게 식별해야 합니다. 메타데이터에 identifier fields를 선언하는 것만으로 데이터 유일성이 보장되는 것은 아닙니다. 키를 신뢰할 수 없거나 중복이 있으면 순변경 로그로 읽을 수 있습니다. 순변경은 같은 전체 행의 INSERT와 DELETE를 상쇄하면서 중복 개수는 보존합니다.

출력에 적용

apply_changelog의 키는 출력에서 매칭할 열입니다. 변환 후 열 이름과 유일성도 충족해야 하며 입력 키가 유일하다는 사실만으로 모든 변환 뒤에도 조건이 유지되지는 않습니다.

필터를 사용하는 경우에는 더 이상 조건을 만족하지 않는 기존 출력 행을 삭제해야 합니다. 그 행의 변경 로그를 버리면 이전 출력이 남습니다. Foundry API에는 해당 변경을 DELETE로 바꾸어 전달하고, 네이티브 MERGE에는 같은 삭제 조건을 작성합니다. 필터된 출력 예제는 NULL 상태도 포함하여 처리합니다.

비증분 빌드

첫 실행 또는 증분 해석 실패 시에는 현재 전체 입력에 동일한 업무 로직을 적용하여 출력을 교체합니다. 전체 이력을 changelog로 다시 생성하고 상관관계를 계산하는 것은 현재 입력을 한 번 읽는 것보다 비효율적일 수 있습니다.

Foundry API로 상태 복제

가상 경로의 원천과 출력이 primary_key로 유일하게 식별된다고 가정합니다. 실제 리소스에 맞게 경로와 키를 변경합니다.

from transforms.api import IncrementalTransformContext, incremental, transform
from transforms.tables import TableInput, TableOutput


@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-foundry-output"),
)
def mirror_foundry_changelog(ctx: IncrementalTransformContext, source, output):
    if ctx.is_incremental:
        output.apply_changelog(source.changelog(["primary_key"]), ["primary_key"])
    else:
        output.write_dataframe(source.dataframe())

네이티브 프로시저로 상태 복제

다음 예제는 NULL이 없는 고유 primary_key와, UPDATE_BEFORE를 제거한 뒤 같은 키 및 _change_ordinal에 액션이 하나뿐이라는 선행 조건을 사용합니다. 같은 순서 값의 여러 액션 중 임의 하나를 고르는 방식은 아닙니다. 조건을 보장할 수 없다면 이 축약을 사용하지 말고 Foundry API가 보장하는 범위 또는 명시적인 업무 순서를 선택합니다.

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

원시 로그의 읽기 범위는 시작 스냅샷을 제외하고 끝 스냅샷을 포함합니다. 끝점은 현재 입력 브랜치의 refs에서 얻으며, 스냅샷 ID의 숫자 크기로 시간순을 추정하지 않습니다.

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-native-output"),
)
def mirror_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.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('primary_key'),
            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 primary_key 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.primary_key = target.primary_key
        WHEN MATCHED AND source._change_type = 'DELETE' THEN DELETE
        WHEN MATCHED AND source._change_type IN ('INSERT', 'UPDATE_AFTER') THEN UPDATE SET *
        WHEN NOT MATCHED AND source._change_type IN ('INSERT', 'UPDATE_AFTER') THEN INSERT *
    """)
    store_watermark(end)

watermark는 출력의 사용자 속성에 저장합니다. NONE은 이전 입력에 스냅샷이 없었던 상태이며 실제 snapshot ID로 프로시저에 보내지 않습니다. 이 경우에는 현재 전체로 초기화합니다. 정상적인 증분 출력에 watermark가 없거나 잘못된 숫자가 있으면 실패시키고 비증분 초기화로 해결합니다.

데이터와 watermark 변경은 Foundry 작업의 트랜잭션 범위 안에서 처리됩니다. 읽어야 할 시작 스냅샷이 만료되었다면 해당 범위를 그대로 복원할 수 있다고 가정하지 않습니다. 보존 정책과 소비 주기를 맞춥니다.

원시 변경 형식, 허용 enum과 실제 Foundry 정리 결과의 차이는 기술 설명에 있습니다.