Python 변환의 증분 계산

증분 변환은 이전 빌드 이력과 이미 계산한 출력을 활용하여 새 입력에 필요한 계산만 수행합니다. 입력과 출력은 서로 다른 데이터셋이어야 합니다. 같은 자산을 양쪽에 지정하면 순환 의존성이 생깁니다.

비증분 변환은 입력 전체를 계산하고 출력 전체를 바꿉니다. 증분 계산이 가능한 실행에서는 새로 추가된 입력을 읽고 그 결과를 기존 출력에 덧붙이는 방식이 기본입니다. @incremental()을 선언해도 모든 빌드가 증분으로 실행되는 것은 아니므로 함수는 스냅샷 실행도 처리해야 합니다.

행별 필터의 계산 범위

가상의 학생 입력에 다음 첫 스냅샷과 추가 배치가 있다고 가정합니다.

hair == "Brown" 필터의 첫 결과는 18번입니다. 다음 증분 빌드는 새 배치에서 20번만 계산하여 추가하므로 최종 출력에는 18번과 20번이 남습니다. 비증분 방식이라면 두 번째 빌드에서도 전체 다섯 행을 다시 필터합니다.

다음 네 구현은 같은 필터를 각 엔진에서 실행합니다. added는 첫 실행이나 스냅샷 실행에서는 전체 입력을, 증분 실행에서는 마지막 처리 이후 새 입력을 나타냅니다. 비증분 구성으로 바꾸려면 @incremental()을 제거하고 읽기 모드도 명시하지 않습니다. 각각 .polars(), .pandas(), .dataframe(), ctx.duckdb().conn으로 읽으면 매번 전체 입력을 계산하고 출력을 교체합니다.

Polars

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


@incremental()
@transform.using(
    students=Input("/Example Project/Inputs/students"),
    processed=Output("/Example Project/Outputs/brown_hair_polars"),
)
def compute(students, processed):
    rows = students.polars(mode="added")
    processed.write_table(rows.filter(pl.col("hair") == "Brown"))

Pandas

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


@incremental()
@transform.using(
    students=Input("/Example Project/Inputs/students"),
    processed=Output("/Example Project/Outputs/brown_hair_pandas"),
)
def compute(students, processed):
    rows = students.pandas(mode="added")
    processed.write_table(rows.loc[rows["hair"] == "Brown"])

DuckDB

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


@incremental()
@transform.using(
    students=Input("/Example Project/Inputs/students"),
    processed=Output("/Example Project/Outputs/brown_hair_duckdb"),
)
def compute(ctx, students, processed):
    connection = ctx.duckdb(read_modes={students: "added"}).conn
    processed.write_table(connection.sql("SELECT * FROM students WHERE hair = 'Brown'"))

PySpark

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


@incremental()
@transform.spark.using(
    students=Input("/Example Project/Inputs/students"),
    processed=Output("/Example Project/Outputs/brown_hair_spark"),
)
def compute(students, processed):
    rows = students.dataframe("added")
    processed.write_dataframe(rows.filter(F.col("hair") == "Brown"))

다른 계산에 적용할 때

열 형식 변경, 날짜 문자열 형식화와 행별 필터는 새 출력이 새 입력에만 의존하므로 단순 추가 방식에 잘 맞습니다. 중복 제거와 집계는 이전 출력도 읽어야 합니다. 한쪽 입력을 전체 참조 표로 읽는 조인은 새 행만 처리할 수 있지만, 양쪽을 added로만 읽으면 앞선 배치와 새 배치 사이의 일치를 놓칠 수 있습니다.

• 설정과 읽기/쓰기 모드는 decorator의 일곱 인자, 증분 실행 조건과 출력 읽기를 설명합니다.
• 스냅샷 참조 입력은 전체 참조 표를 읽는 네 엔진의 조인을 설명합니다.
• 추가, 중복 제거와 집계는 이전 출력을 사용하는 계산과 스키마 진화를 다룹니다.
• 두 입력의 증분 조인은 단조 증가 식별자라는 조건이 있는 처리 범위를 설명합니다.
• 이력 데이터셋과 파일 수에 따른 병합은 스냅샷 누적의 위험과 작은 파일 문제를 다룹니다.
• 트랜잭션 배치 제한은 한 작업이 읽을 범위를 나누고 출력을 최신으로 유지하는 방법입니다.
• 증분 오류와 브랜치는 처리 범위와 현재 데이터셋 뷰가 어긋난 상황을 설명합니다.

미리보기와 빈 스트리밍 구간

Code Repositories의 Preview는 require_incremental=True를 지정해도 비증분으로 실행됩니다. VS Code preview는 Spark와 lightweight의 v1/v2 증분 동작을 지원하므로 실제 증분 범위를 확인할 때 사용합니다.

스트리밍 데이터셋의 새 증분 구간에 파일이 없으면 Polars lazy 읽기에서 BinderException이 발생할 수 있습니다. ctx.is_incremental과 eager 입력의 빈 상태를 먼저 검사하는 lazy 입력 예제를 적용하거나 구간에 파일이 있을 때 실행합니다.

미디어 집합 입력도 증분 계산을 지원합니다. 미디어 항목을 읽는 증분 처리와 다시 미디어 집합으로 쓰는 출력의 지원 조건은 따로 확인해야 합니다. 배치 크기 제한 역시 자산 유형별 조건을 따릅니다.