출력을 바꾸지 않고 트랜잭션 중단하기

ctx.abort_job()은 변환 작업을 성공으로 끝내면서 출력 트랜잭션을 커밋하지 않습니다. 출력 데이터셋은 이전 상태를 유지합니다. 실행 결과를 갱신할 필요가 없을 때 빌드를 일부러 실패시키지 않고 사용하는 방법입니다.

출력이 커밋되면 하위 데이터셋은 stale 상태가 되고 다음 수동 빌드나 예약 빌드에서 다시 계산됩니다. 입력 내용에 따라 갱신을 제한하거나, 입력 데이터셋 없이 외부 API를 조회하느라 강제 빌드가 필요하거나, 매번 갱신되는 writeback 데이터셋의 결과가 실질적으로 같을 때 중단 조건을 둘 수 있습니다. 불필요한 하위 재계산을 줄이면서 실제 실패를 구분합니다.

트랜잭션 이력에서는 중단된 작업이 회색의 성공한 작업으로 표시됩니다.

오늘의 데이터가 도착했을 때만 쓰기

다음 네 예제는 trip_date 날짜 열이 있는 가상의 여행 데이터셋을 사용합니다. 오늘의 여행이 없으면 기존 출력을 유지하고, 있으면 해당 행을 출력합니다. date.today()는 실행 환경의 날짜를 사용하므로 업무상 날짜가 특정 시간대를 따라야 하면 그 시간대를 기준으로 날짜를 계산합니다.

Polars

from datetime import date

import polars as pl
from transforms.api import Input, Output, transform


@transform.using(
    trips=Input("/Example Project/Inputs/trips"),
    processed=Output("/Example Project/Outputs/trips_processed"),
)
def compute(ctx, trips, processed):
    today = trips.polars().filter(pl.col("trip_date") == date.today())
    if today.is_empty():
        ctx.abort_job()
    else:
        processed.write_table(today)

DuckDB

from datetime import date

from transforms.api import Input, Output, transform


@transform.using(
    trips=Input("/Example Project/Inputs/trips"),
    processed=Output("/Example Project/Outputs/trips_processed"),
)
def compute(ctx, trips, processed):
    connection = ctx.duckdb().conn
    today = connection.sql(
        "SELECT * FROM trips WHERE trip_date = ?", params=[date.today()]
    ).fetchdf()
    if today.empty:
        ctx.abort_job()
    else:
        processed.write_table(today)

fetchdf()는 Pandas DataFrame으로 결과 전체를 구체화하므로 해당 결과가 메모리에 들어가는 크기여야 합니다.

Pandas

from datetime import date

import pandas as pd
from transforms.api import Input, Output, transform


@transform.using(
    trips=Input("/Example Project/Inputs/trips"),
    processed=Output("/Example Project/Outputs/trips_processed"),
)
def compute(ctx, trips, processed):
    rows = trips.pandas()
    trip_dates = pd.to_datetime(rows["trip_date"]).dt.date
    today = rows.loc[trip_dates == date.today()]
    if today.empty:
        ctx.abort_job()
    else:
        processed.write_table(today)

PySpark

from datetime import date

from pyspark.sql import functions as F
from transforms.api import Input, Output, transform


@transform.spark.using(
    trips=Input("/Example Project/Inputs/trips"),
    processed=Output("/Example Project/Outputs/trips_processed"),
)
def compute(ctx, trips, processed):
    today = trips.dataframe().filter(F.col("trip_date") == F.lit(date.today()))
    if not today.head(1):
        ctx.abort_job()
    else:
        processed.write_dataframe(today)

존재 여부만 필요하면 Spark의 전체 행 수를 count()하는 대신 head(1)로 한 행만 확인하는 편이 보통 더 적은 작업을 수행합니다.

ignored와 aborted의 차이

아무것도 쓰지 않고 정상 반환하는 경로를 명시적 중단과 혼동하지 않습니다. 기존 출력을 유지하려는 조건에서는 중단 API로 의도를 지정합니다.

증분 처리의 입력 범위

증분 계산은 입력과 출력의 커밋된 트랜잭션만 사용합니다. 작업 전체를 중단하면 다음 빌드는 중단된 트랜잭션이 없었던 것처럼 입력 범위를 읽고 재처리하므로 증분 실행을 계속할 수 있습니다.

PySpark에서는 개별 출력의 abort()도 가능하지만, 여러 출력 중 일부만 중단하면 출력마다 마지막 입력 트랜잭션 범위가 달라집니다. 다음 빌드는 Provenance records for the previous build are inconsistent 검사 실패로 스냅샷 실행을 선택합니다. require_incremental=True이면 InconsistentProvenanceRecords 오류로 실패합니다.

다중 출력 증분 변환에서는 ctx.abort_job()으로 모든 출력을 함께 중단합니다. v2_semantics에서는 이후 빌드의 증분 실행을 유지하면서 출력을 중단할 수 있는 방법이 작업 전체 중단뿐입니다.