증분 decorator와 읽기/쓰기 모드

@incremental()은 이전 빌드 이력을 확인하여 현재 빌드가 증분으로 실행될 수 있는지 결정하고, 입력과 출력 및 실행 문맥에 증분 기능을 제공합니다. 여기서 증분/비증분 실행은 모두 decorator를 선언한 변환 안에서 실제로 선택된 실행 방식을 뜻합니다.

Spark 경로에서는 TransformInput, TransformOutput, TransformContext 대신 각각 IncrementalTransformInput, IncrementalTransformOutput, IncrementalTransformContext의 기능을 사용합니다. Lightweight 경로에서도 각 엔진의 입력과 출력 메서드 및 ctx.is_incremental을 사용합니다.

일곱 설정 인자

transforms.api.incremental(
    require_incremental=False,
    semantic_version=1,
    snapshot_inputs=None,
    allow_retention=False,
    allow_user_deletes=False,
    strict_append=False,
    v2_semantics=False,
)

위 코드는 API 서명이며 실제 사용에서는 from transforms.api import incremental로 가져와 decorator로 적용합니다.

인자	기본값	의미와 조건
`require_incremental`	`False`	증분 실행이 불가능하면 자동 스냅샷 대신 실패시킴. 이전에 빌드하지 않은 출력이 있거나 semantic version 변경으로 스냅샷을 요청한 경우는 예외
`semantic_version`	`1`	이전 실행과 값이 달라지면 다음 실행을 비증분으로 수행. 이전 기록에 값이 없으면 1로 간주
`snapshot_inputs`	`None`	갱신이나 삭제가 가능한 전체 참조 입력의 이름 목록
`allow_retention`	`False`	입력과 출력에 적용한 보존 정책의 파일 삭제를 허용하면서 증분 처리 유지
`allow_user_deletes`	`False`	보존 정책 밖의 입력 파일 삭제를 증분 가능성 검사에서 허용. `v2_semantics=True`가 필요하며 lightweight는 지원하지 않음
`strict_append`	`False`	증분 입력을 처리하는 실행의 출력 트랜잭션을 `APPEND`로 만들어 기존 파일 덮어쓰기 금지
`v2_semantics`	`False`	v2 증분 의미론 사용. 권장 설정이며 비 Catalog 자산을 증분으로 읽거나 쓰려면 필요

require_incremental=True여도 최초 출력 생성과 명시적인 semantic version 변경은 스냅샷 실행을 허용합니다. 원인은 드라이버 로그의 transforms.api._incremental: Not running incrementally 경고에서 확인합니다.

기존 변환을 증분으로 전환할 때 이전 semantic version이 없으면 1로 취급하므로, 단순히 기본값으로 decorator를 추가했다는 이유만으로 새 스냅샷을 강제하지 않습니다. 전체 재계산이 필요하면 값을 변경합니다. 기존 증분 변환에서 v2_semantics를 활성화하면 다음 빌드가 한 번 SNAPSHOT으로 실행되는 영향도 고려합니다.

strict_append=True일 때 덮어쓰기 금지는 데이터 파일뿐 아니라 Parquet 요약 메타데이터와 Hadoop SUCCESS 파일에도 적용합니다. 입력을 증분으로 처리할 수 없으면 이 설정만으로는 막지 못하고 스냅샷으로 실행되어 기존 파일 교체가 가능합니다. 스냅샷 대체를 제한하려면 require_incremental=True도 사용하되 앞서 설명한 예외는 남습니다.

입력 읽기 모드

트랜잭션 배치 제한이 없는 일반 입력의 모드는 다음과 같습니다. 같은 모드가 표 읽기와 filesystem() 읽기에 적용됩니다.

모드	증분 실행	비증분 실행
`added` 기본값	지난 처리 이후 추가된 데이터	전체 입력, 모든 데이터를 아직 읽지 않은 것으로 취급
`previous`	지난 실행에 제공한 전체 입력	빈 입력
`current`	이번 실행의 전체 입력	전체 입력, `added`와 같음

Polars와 Pandas는 input.polars(mode="added"), input.pandas(mode="added")처럼 읽습니다. Spark는 input.dataframe("added")를 사용하고 DuckDB는 ctx.duckdb(read_modes={input: "added"})로 입력별 모드를 지정합니다. 기본값을 쓰면 모드 인자를 생략할 수 있습니다.

스냅샷 참조 입력은 별도 규칙을 따릅니다. added와 current가 항상 전체 입력을 반환하고 다른 모드는 빈 입력을 반환합니다. 배치 제한을 적용한 일반 입력의 current는 전체 최신 입력 대신 현재 배치까지의 뷰를 나타냅니다.

입력 뷰를 만들 때 마지막 SNAPSHOT부터 누적된 트랜잭션을 읽으므로 이력이 매우 길면 실행이 점점 느려질 수 있습니다. 해당 입력의 스냅샷 빌드를 검토하되 하위 증분 처리와 이력 보존에 미치는 영향을 함께 확인합니다.

출력 쓰기 모드

모드	동작	기본으로 선택되는 실행
`modify`	기존 출력에 이번 쓰기를 반영. 표 쓰기는 새 DataFrame을 추가	증분 실행
`replace`	이번에 쓴 데이터로 출력 전체 교체	비증분 실행

새 입력을 읽는 added와 증분 출력의 modify 조합으로 행별 필터를 처음 실행하거나 이어 실행하는 코드를 함께 처리할 수 있습니다. 집계 결과처럼 전체 출력 상태를 다시 써야 할 때는 쓰기 전에 output.set_mode("replace")를 호출합니다.

transform(), transform.using()과 해당 Spark 선언은 핸들을 직접 받아 읽기/쓰기 모드를 지정할 수 있습니다. transform_df()와 transform_pandas()는 반환값을 decorator가 저장하므로 기본 모드만 사용할 수 있습니다. 이전 출력이 필요한 조인, 집계나 중복 제거에는 핸들을 받는 형식을 사용합니다.

직접 지정한 모드가 증분과 비증분 양쪽에서 올바른지 확인합니다. 실행별 처리가 다르면 ctx.is_incremental로 나눕니다.

이전 출력과 이번에 쓴 출력 읽기

출력의 기본 읽기 모드는 current이지만 이전 실행의 결과를 읽으려는 일반적인 목적에는 previous를 사용합니다. Lightweight 출력은 output.pandas(mode="previous")처럼 모드를 키워드 인자로 지정합니다.

출력 읽기	출력 쓰기	이번 쓰기	읽는 내용
`current`	`modify`	아직 없음	이전 결과가 목적이면 `previous` 사용
`current`	`modify`	있음	빌드 시작 시 출력과 이번 쓰기를 합한 전체
`current`	`replace`	아직 없음	이전 결과가 목적이면 `previous` 사용
`current`	`replace`	있음	이번 빌드에서 쓴 데이터
`added`	둘 다	아직 없음	이전 결과가 목적이면 `previous` 사용
`added`	둘 다	있음	이번 빌드에서 쓴 데이터
`previous`	`modify`	무관	빌드 시작 시 출력
`previous`	`replace`	무관	빌드 시작 시 출력

비증분 실행에서는 이전 출력 이력을 사용하지 않으므로 previous가 빈 결과를 반환합니다. Spark에서 이 빈 DataFrame을 만들려면 schema를 전달합니다. 증분 실행에서 schema를 전달하면 실제 이전 출력과 열 자료형, null 허용 여부 및 열 순서가 같은지 검사하며 다르면 오류가 발생합니다.

Foundry는 저장된 모든 열을 nullable로 취급합니다. previous를 읽을 schema에 nullable=False가 있으면 SchemaMismatchError가 날 수 있으므로 각 열을 nullable로 선언합니다. 출력도 df.select(schema.fieldNames())처럼 같은 열 순서로 씁니다.

previous는 직전 실행에서 추가한 배치만을 뜻하지 않습니다. 빌드 시작 시의 전체 출력입니다. 직전 실행의 배치만 따로 필요하면 보조 출력을 만들고 매번 replace로 해당 배치나 관련 메타데이터를 저장합니다.

현재 실행에서 쓴 데이터를 검사하려면 쓰기 후 added로 읽는 편이 의도가 분명합니다. 전체 출력 검사를 위해 Spark 계산을 다시 수행하거나 캐시할 필요를 줄일 수 있습니다.

증분 실행을 유지하는 조건

일반 입력의 파일 변경과 출력의 이력이 다음 조건을 만족해야 합니다.

일반 입력에는 파일이 추가되었거나, 명시적으로 허용한 보존 정책 삭제/사용자 삭제만 있어야 합니다.
UPDATE가 기존 파일을 수정하거나 덮어쓰면 증분 조건을 깨뜨립니다. 파일만 추가한 UPDATE는 일반 증분 입력으로 처리할 수 있습니다. 트랜잭션 배치 제한은 더 엄격하여 UPDATE를 허용하지 않습니다.
SNAPSHOT으로 전체 입력을 바꾸면 일반 입력의 연속성이 끊깁니다. 전체 참조 입력으로 다뤄도 계산이 맞는 경우에만 snapshot_inputs로 선언합니다.
일반 입력의 시작 트랜잭션은 이전 실행과 일관되어야 합니다.
다중 출력에서는 이전 빌드가 있는 모든 출력의 마지막 커밋이 같은 변환에서 생성되어야 합니다. 빌드 이력이 없는 새 출력은 이 검사에서 제외합니다.
변환에는 입력이 하나 이상 있어야 하며, 입력 자산과 출력 자산이 같아서는 안 됩니다.

조건이 깨지면 기본적으로 비증분 실행을 선택하고 입력을 전체로 제시하며 출력은 교체합니다. 이전 출력 읽기는 빈 결과가 됩니다. require_incremental=True라면 최초 생성과 semantic version 변경 예외를 제외하고 실패합니다.

입력이나 스냅샷 참조 입력을 추가하거나 제거하는 것 자체는 증분을 유지할 수 있습니다. 입력 구성이 바뀌어 계산 의미가 달라졌다면 기존 출력도 재계산해야 하는지 판단하고 semantic version을 조정합니다.

삭제 허용 범위

allow_retention=True는 보존 정책이 만든 removed 파일 변경만 허용합니다. 보존 정책 밖의 삭제나 파일 수정까지 허용하지 않습니다.

allow_user_deletes=True는 보존 정책 밖의 입력 파일 삭제만 허용합니다. 보존 정책 삭제도 허용하려면 두 인자를 함께 켭니다. 이 기능은 Spark에서 v2_semantics=True와 함께 사용하며 lightweight에는 적용할 수 없습니다.

삭제를 허용한다는 것은 증분 가능성 검사에서 해당 삭제를 받아들인다는 뜻입니다. 이미 계산한 출력에서 그 입력의 기여분이 자동으로 제거되는 기능은 아닙니다. 입력 삭제 후에도 기존 출력이 올바른 계산에만 적용합니다.

기초 필터 예제의 네 엔진에 @incremental(allow_retention=True)를 적용하면 추가 파일과 보존 정책 삭제가 함께 있는 입력을 계속 처리할 수 있습니다. 사용자 삭제가 필요한 Spark 구성은 @incremental(v2_semantics=True, allow_user_deletes=True)를 사용하고, 두 종류의 삭제를 모두 허용하려면 여기에 allow_retention=True를 추가합니다.

Code Repositories Preview는 항상 비증분이며, VS Code preview는 Spark와 lightweight의 v1/v2 증분 실행을 지원합니다. 미리보기 종류를 구분하여 읽기 범위와 출력 누적을 확인합니다.

Keyboard shortcuts