증분 입력의 트랜잭션 배치 제한

기본 증분 작업은 입력마다 아직 처리하지 않은 트랜잭션 전체를 한 번에 처리합니다. semantic version 변경으로 처음부터 재계산하거나 오랫동안 쌓인 입력을 따라잡을 때 한 작업의 범위가 커질 수 있습니다. Input(..., transaction_limit=...)는 작업마다 읽을 트랜잭션 수를 제한합니다. 행 수나 바이트 수의 상한은 아니므로 큰 트랜잭션 하나의 크기도 따로 고려합니다.

지원 조건

이 장은 transactional dataset 입력의 배치 제한을 다룹니다.

• @incremental(v2_semantics=True)를 사용합니다.
• Python transforms 3.25.0 이상이 필요합니다. 특정 버전을 사용해야 하면 모듈 고정을 구성합니다.
• Lightweight 변환에는 이 기능을 사용할 수 없습니다.
• 제한할 입력은 스냅샷 참조 입력으로 지정하지 않습니다.
• 입력의 현재 뷰에는 시작 트랜잭션으로 가능한 SNAPSHOT을 제외하고 APPEND만 있어야 합니다.

현재 뷰에 DELETE나 UPDATE가 하나라도 있으면 Build2:InvalidTransactionTypeForBatchInputResolution으로 실패합니다. 기존 파일을 바꾸지 않고 파일만 추가한 UPDATE라도 배치 제한의 조건을 충족하지 않습니다. 이는 일반 증분 계산보다 엄격한 조건입니다.

기존 증분 변환에서 v2를 새로 켜면 다음 빌드가 한 번 SNAPSHOT으로 실행됩니다.

입력별로 다른 제한 지정하기

다음 가상 예제의 세 일반 입력은 category를 포함한 동일한 schema를 가지며, 스냅샷 참조 표에는 category별 label이 한 행씩 있습니다. 새 행을 합쳐 현재 참조 정보와 조인합니다. 예제는 Spark의 기존 @transform 선언을 사용합니다.

from transforms.api import Input, Output, incremental, transform


@incremental(
    v2_semantics=True,
    strict_append=True,
    snapshot_inputs=["snapshot_input"],
)
@transform(
    input_1=Input("/Example Project/Inputs/first", transaction_limit=3),
    input_2=Input("/Example Project/Inputs/second", transaction_limit=2),
    input_3=Input("/Example Project/Inputs/third"),
    snapshot_input=Input("/Example Project/Inputs/categories"),
    output=Output("/Example Project/Outputs/combined"),
)
def compute(input_1, input_2, input_3, snapshot_input, output):
    added = (
        input_1.dataframe()
        .unionByName(input_2.dataframe())
        .unionByName(input_3.dataframe())
    )
    categories = snapshot_input.dataframe()
    output.write_dataframe(added.join(categories, "category", "left"))

input_1은 최대 3개, input_2는 최대 2개의 트랜잭션을 읽습니다. input_3은 제한 없이 미처리 범위를 읽고 참조 입력은 매번 전체 뷰를 읽습니다. 여러 입력에 같은 제한을 부여해도 입력별 트랜잭션의 시간 범위가 서로 일치하는 것은 아닙니다.

added, previous, current가 가리키는 범위

예제의 트랜잭션 A~C를 처리한 뒤 D~J가 새로 도착했다고 가정합니다. 제한이 없으면 다음 한 작업이 added=D~J, previous=A~C, current=A~J를 읽습니다.

3개 제한을 설정하면 세 작업으로 따라잡습니다.

current도 입력의 최신 HEAD 전체가 아니라 해당 작업에서 처리할 배치까지의 범위입니다.

semantic version 변경 등으로 스냅샷 재계산을 시작하면 이전 처리 경계를 초기화합니다. 제한이 없을 때는 added=A~J, previous=없음, current=A~J입니다. 3개 제한이 있으면 다음 네 작업이 필요합니다.

최신 상태까지 계속 빌드하는 스케줄

성공한 빌드가 한 배치만 처리했으면 출력이 여전히 stale일 수 있습니다. 다음과 같이 스케줄을 구성합니다.

1. Data Lineage의 오른쪽 패널에서 Manage schedule을 열고 Create new schedule을 선택합니다.
2. 배치 제한을 설정한 변환의 출력 데이터셋을 target resource로 지정합니다.
3. When to build에서 특정 자산 갱신, 특정 시각, 수동 트리거 또는 여러 시간/이벤트 조건 중 필요한 방식을 정하고 빈도를 설정합니다.
4. 아래의 Advanced options를 펼칩니다.
5. Re-trigger schedule upon completion of a successful build를 켜고 Save로 저장합니다.

이 옵션은 출력이 최신 상태가 될 때까지만 재실행합니다. Force build와 동시에 사용할 수 없습니다. Force build는 최신 여부와 무관하게 실행하므로 두 설정을 함께 적용하면 종료 조건이 사라지기 때문입니다. 새 입력이 빠르게 계속 도착하면 재빌드도 계속될 수 있습니다.

재실행 옵션은 스케줄의 대상 자산 중 하나 이상이 트랜잭션 배치 제한을 사용할 때만 표시됩니다.

작업이 실제로 읽은 범위 확인하기

해당 작업의 Spark details에서 Snapshot/Incremental 탭을 엽니다.

이전 작업에 제한이 없었거나 마지막 배치까지 처리했다면 Last read transaction은 Previous end transaction과 같습니다. 제한이 있고 아직 마지막 배치가 아니었다면 실제 처리 경계가 이전 뷰의 끝보다 앞에 있습니다.

트랜잭션을 선택하면 입력 데이터셋의 History로 이동하며 해당 트랜잭션이 강조됩니다. 증분 설정과 출력 모드를 함께 확인하여 배치 제한 때문에 범위가 줄어든 경우와 증분 조건 자체가 깨진 경우를 구분합니다.