Iceberg 변경 로그의 키, 순변경과 순서
변경 로그는 두 스냅샷 사이의 삽입, 수정과 삭제를 표현합니다. 신뢰할 수 있는 고유 식별자가 있으면 식별자 로그를, 키를 신뢰할 수 없거나 중복을 보존해야 하면 순변경(Net changes) 로그를 선택합니다.
원시 Iceberg 변경 형식, API 자료형이 허용하는 enum과 Foundry 메서드가 실제로 정리하여 반환하는 결과는 구분해야 합니다.
세 가지 계약
| 구분 | 변경 형태 |
|---|---|
| 네이티브 identifier changelog | compute_updates로 같은 키의 DELETE/INSERT를 UPDATE_BEFORE/UPDATE_AFTER 쌍으로 표현합니다. 여러 스냅샷의 중간 변경도 포함합니다. |
Foundry source.changelog(keys) | UPDATE_BEFORE를 제거하고 고유 식별자별 최신 액션을 선택합니다. 원천 식별자의 유일성을 검증합니다. |
| Net changes | 모든 데이터 열로 행을 비교하여 INSERT/DELETE를 상쇄합니다. 동일 행의 중복 개수는 보존합니다. |
변경 유형의 일반 enum에는 INSERT, DELETE, UPDATE_BEFORE와 UPDATE_AFTER가 있습니다. 이 enum에 UPDATE_BEFORE가 있다는 사실만으로 현재 Foundry 식별자 메서드가 그 행을 반환한다고 판단하지 않습니다.
반환 열과 변경 유형
데이터 열에 변경 메타데이터가 추가됩니다.
| 열 | 유형과 의미 |
|---|---|
| 원래 데이터 열 | 원천의 열과 자료형 |
_change_type | 문자열 변경 유형 |
_change_ordinal | 변경 순서를 나타내는 정수. 같은 스냅샷의 여러 행에 같은 값이 있을 수 있습니다. |
_commit_snapshot_id | 변경이 발생한 스냅샷 ID |
스냅샷 ID는 식별자이며 숫자의 대소로 시간순을 판단하지 않습니다. 아래 설명 표의 ordinal1/2/3은 예시 이력의 상대 순서를 나타내는 값입니다. 실제 처리에서는 해당 변경 구간이 제공한 순서를 사용하며 표의 숫자를 고정된 전역 커서로 사용하지 않습니다.
| 변경 유형 | 의미 |
|---|---|
INSERT | 새 행 또는 새 식별자의 행 |
DELETE | 제거할 행 |
UPDATE_BEFORE | 네이티브 식별자 변경에서 갱신 전 이미지 |
UPDATE_AFTER | 갱신 후 이미지 |
식별자 로그
하나 이상의 열이 원천 행을 고유하게 식별해야 합니다. 신뢰할 수 있는 키가 있으면 전체 행으로 순변경을 계산하는 것보다 대체로 효율적이며 갱신 의미도 사용할 수 있습니다.
Foundry 메서드는 source.changelog(["id"])처럼 호출합니다. Iceberg 메타데이터의 identifier fields는 유일성을 강제하지 않으므로 그 선언만으로 API의 조건을 충족했다고 판단하지 않습니다. Transforms는 해당 메타데이터 선언에만 의존하지 않습니다.
네이티브 변경 로그의 간단한 예는 다음과 같습니다.
| id | value | _change_type | _change_ordinal |
|---|---|---|---|
| id1 | val1 | UPDATE_BEFORE | 1 |
| id1 | val2 | UPDATE_AFTER | 1 |
| id2 | val2 | INSERT | 2 |
같은 변화에 대해 Foundry 식별자 메서드가 반환하는 정리 결과는 다음과 같습니다. 이전 이미지는 제외됩니다.
| id | value | _change_type | _change_ordinal |
|---|---|---|---|
| id1 | val2 | UPDATE_AFTER | 1 |
| id2 | val2 | INSERT | 2 |
순변경 로그
식별자를 생략하여 source.changelog()를 호출합니다. 네이티브 프로시저는 identifier_columns를 생략하고 net_changes를 true로 설정합니다.
원시 행 수정은 삭제와 삽입으로 표현할 수 있습니다. 순변경은 키가 아니라 모든 데이터 열을 비교하므로 값이 1에서2, 다시3으로 바뀌면 중간값2의 삽입/삭제가 상쇄되고 이전값1의 DELETE와 최종값3의 INSERT가 남습니다.
| id | value | _change_type | _change_ordinal |
|---|---|---|---|
| id1 | val1 | DELETE | 1 |
| id1 | val2 | INSERT | 1 |
순변경은 INSERT와 DELETE만 사용합니다. 중복 행을 하나로 만드는 distinct 연산이 아니며 같은 값의 행이 두 개 늘었다면 그 개수도 남아야 합니다.
V0부터 V3까지의 테이블 변화
다음은 가상 값으로 구성한 네 상태입니다. V3에서는 id2가 없어지고 id1/val3 행이 두 개가 됩니다.
| V0 | V1 | V2 | V3 |
|---|---|---|---|
| [id1, val1] | [id1, val2] | [id1, val3] | [id1, val3] |
| [id2, val2] | [id1, val3] |
식별자 로그: V0에서 V1
원시 Iceberg의 전후 이미지:
| id | value | _change_type | _change_ordinal |
|---|---|---|---|
| id1 | val1 | UPDATE_BEFORE | 1 |
| id1 | val2 | UPDATE_AFTER | 1 |
Foundry 메서드의 결과:
| id | value | _change_type | _change_ordinal |
|---|---|---|---|
| id1 | val2 | UPDATE_AFTER | 1 |
식별자 로그: V1에서 V2
원시 전후 이미지와 삽입:
| id | value | _change_type | _change_ordinal |
|---|---|---|---|
| id1 | val2 | UPDATE_BEFORE | 2 |
| id1 | val3 | UPDATE_AFTER | 2 |
| id2 | val2 | INSERT | 2 |
Foundry 메서드의 결과:
| id | value | _change_type | _change_ordinal |
|---|---|---|---|
| id1 | val3 | UPDATE_AFTER | 2 |
| id2 | val2 | INSERT | 2 |
식별자 로그: V0에서 V2
네이티브 프로시저는 중간 스냅샷의 변화도 포함합니다.
| id | value | _change_type | _change_ordinal |
|---|---|---|---|
| id1 | val1 | UPDATE_BEFORE | 1 |
| id1 | val2 | UPDATE_AFTER | 1 |
| id1 | val2 | UPDATE_BEFORE | 2 |
| id1 | val3 | UPDATE_AFTER | 2 |
| id2 | val2 | INSERT | 2 |
Foundry는 before 이미지를 제거하고 id별 최신 액션을 선택합니다.
| id | value | _change_type | _change_ordinal |
|---|---|---|---|
| id1 | val3 | UPDATE_AFTER | 2 |
| id2 | val2 | INSERT | 2 |
구간의 시작값과 끝값을 설명할 때 id1이 val1에서 val3으로 바뀌었다고 말할 수 있지만, 그 설명을 Foundry 반환값의 before/after 쌍으로 해석하지 않습니다.
식별자 로그: V0에서 V3
V3에는 같은 id1 행이 중복되므로 고유 식별자 조건이 성립하지 않습니다. Foundry 식별자 changelog 생성은 실패합니다. 키 정의가 유효하지 않은 데이터를 임의로 한 행으로 축약하여 정상적인 키 기반 처리인 것처럼 만들지 않습니다.
순변경의 구간별 결과
V0에서 V1
| id | value | _change_type | _change_ordinal |
|---|---|---|---|
| id1 | val1 | DELETE | 1 |
| id1 | val2 | INSERT | 1 |
V1에서 V2
| id | value | _change_type | _change_ordinal |
|---|---|---|---|
| id1 | val2 | DELETE | 2 |
| id1 | val3 | INSERT | 2 |
| id2 | val2 | INSERT | 2 |
V0에서 V2
상쇄 전의 변화는 다음과 같습니다.
| id | value | _change_type | _change_ordinal |
|---|---|---|---|
| id1 | val1 | DELETE | 1 |
| id1 | val2 | INSERT | 1 |
| id1 | val2 | DELETE | 2 |
| id1 | val3 | INSERT | 2 |
| id2 | val2 | INSERT | 2 |
id1/val2의 INSERT와 DELETE가 상쇄되어 다음이 남습니다.
| id | value | _change_type | _change_ordinal |
|---|---|---|---|
| id1 | val1 | DELETE | 1 |
| id1 | val3 | INSERT | 2 |
| id2 | val2 | INSERT | 2 |
V0에서 V3
전체 변화는 다음과 같습니다.
| id | value | _change_type | _change_ordinal |
|---|---|---|---|
| id1 | val1 | DELETE | 1 |
| id1 | val2 | INSERT | 1 |
| id1 | val2 | DELETE | 2 |
| id1 | val3 | INSERT | 2 |
| id2 | val2 | INSERT | 2 |
| id2 | val2 | DELETE | 3 |
| id1 | val3 | INSERT | 3 |
id1/val2와 id2/val2 각각의 삽입/삭제가 상쇄됩니다. id1/val3의 삽입 두 개는 서로 상쇄되지 않습니다.
| id | value | _change_type | _change_ordinal |
|---|---|---|---|
| id1 | val1 | DELETE | 1 |
| id1 | val3 | INSERT | 2 |
| id1 | val3 | INSERT | 3 |
유일성 검사 위치
원천 식별자
source.changelog(["id"])는 원천이 해당 열로 유일하게 식별되어야 합니다. 중복이면 다음과 같은 오류가 발생합니다.
ValueError: Duplicate rows found for identifier columns ['id'] in changelog view.
키를 보장할 수 없다면 식별자 없는 순변경 읽기를 사용합니다. 순변경을 읽을 수 있다는 사실이 같은 데이터를 키 기반 upsert API에 그대로 적용할 수 있다는 뜻은 아닙니다.
변경 DataFrame과 대상 테이블
output.apply_changelog(changes, ["id"])의 id는 출력에서 매칭할 키입니다. 변경 DataFrame은 키당 최대 하나의 액션이어야 하고, 출력 테이블도 같은 키로 유일해야 합니다.
Exception: apply_changelog failed: duplicate keys detected in changelog dataframe for identifier columns ['id']
Exception: apply_changelog failed: multiple rows in destination table match key ['id']
변경 DataFrame만 줄였다고 이미 중복된 대상 테이블이 고쳐지지는 않습니다. 원천, 변환 뒤 변경 액션 및 대상의 조건을 각각 확인합니다.
최신 액션을 고를 수 있는 조건
Foundry의 식별자 메서드가 이미 정리한 결과에는 같은 정리를 반복할 필요가 없습니다. 네이티브 원시 식별자 로그 또는 사용자 변환 뒤 여러 시점의 액션을 정리할 때에는 순서가 보장되어야 합니다.
다음 함수는 UPDATE_BEFORE를 제거한 뒤 같은 키와 ordinal에 액션이 하나뿐이라는 선행 조건에서만 사용합니다. 키가 유일하다는 선언만으로 이 순서 조건이 자동 성립하지는 않습니다. 같은 ordinal에 여러 액션이 있다면 임의 row_number를 적용하지 말고 실제 업무상 순서를 정하거나 해당 보장을 제공하는 API를 사용합니다. 입력에는 함수가 임시로 사용하는 _latest_action_row 열을 두지 않습니다.
from pyspark.sql import functions as F
from pyspark.sql.window import Window
def latest_native_identifier_actions(changes, identifier_columns):
after_images = changes.filter(F.col("_change_type") != "UPDATE_BEFORE")
ordering = Window.partitionBy(*identifier_columns).orderBy(F.col("_change_ordinal").desc())
return (
after_images
.withColumn("_latest_action_row", F.row_number().over(ordering))
.filter(F.col("_latest_action_row") == 1)
.drop("_latest_action_row")
)
이 함수는 순변경에서 합법적인 중복 개수를 제거하는 용도가 아닙니다. 신뢰할 수 없는 키로 net changes를 한 행씩만 남기면 실제 데이터가 손실될 수 있습니다. 키 없는 순변경과 유일한 대상 키를 요구하는 apply_changelog의 경계를 유지합니다.
네이티브 프로시저의 주요 인수
| 인수 | 의미 |
|---|---|
table | 필수 원천 테이블 이름 |
changelog_view | 생성할 뷰 이름 |
identifier_columns | 전후 이미지를 계산할 식별자 열 목록 |
compute_updates | identifier_columns를 주면 기본 true, 그렇지 않으면 기본 false입니다. true인데 열 목록을 생략하면 테이블의 identifier fields를 사용할 수 있습니다. |
net_changes | 기본 false입니다. compute_updates가 true일 때는 false여야 합니다. |
options | 스냅샷 ID 또는 시각으로 읽을 구간 지정 |
start-snapshot-id 및 start-timestamp는 시작점을 제외하며, 생략하면 첫 스냅샷부터 포함합니다. end-snapshot-id 및 end-timestamp는 끝점을 포함하며 기본값은 현재 스냅샷입니다. 실행하면 변경 뷰가 생성되고 뷰 이름이 반환됩니다. 데이터 행은 그 뷰를 질의하여 읽습니다.
Carry-over 행은 기본적으로 제거되며, 데이터 파일의 재작성만으로 논리 변경이 생긴 것처럼 처리하지 않습니다. Changelog 소비 테이블은 copy-on-write로 유지합니다. Merge-on-read가 만든 delete file이 읽기 구간에 있으면 지원되지 않습니다.