스케줄 매개변수화

스케줄 매개변수화는 같은 Python 변환을 서로 다른 값으로 실행하는 Beta 기능입니다. 플랫폼 인스턴스에 따라 제공되지 않을 수 있고, 현재 Python 변환만 지원합니다.

저장소 버전과 매개변수 선언

다음 최소 버전을 충족해야 합니다.

버전이 낮으면 저장소 업그레이드로 구성을 갱신합니다. 매개변수는 이름, 자료형과 기본값으로 정의합니다. 이름은 Python 변환 선언의 변수명이고, 자료형은 StringParam, IntegerParam, FloatParam, BooleanParam 중 하나입니다. 스케줄에서 값을 전달하지 않으면 변환에 정의한 기본값을 사용합니다.

다음은 town_name 문자열 열이 있는 지역 데이터셋에서 선택한 지역의 행을 남기고 설정값 risk_factor를 출력 열로 기록하는 예제입니다. 저장소에 Polars가 필요하며 경로는 실제 데이터셋으로 지정합니다. 기본값은 Seattle과 5입니다.

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


@transform.using(
    processed_towns=Output("/examples/towns/parameterized"),
    towns=Input("/examples/towns/input"),
    town_name=StringParam("Seattle"),
    risk_factor=IntegerParam(5),
)
def select_town(processed_towns, towns, town_name, risk_factor):
    selected_town_name = town_name.value
    selected_risk_factor = risk_factor.value
    selected = towns.polars().filter(
        pl.col("town_name") == selected_town_name
    )
    processed_towns.write_table(
        selected.with_columns(pl.lit(selected_risk_factor).alias("risk_factor"))
    )

코드는 매개변수 객체의 .value에서 실제 값을 읽습니다.

스케줄에서 활성화

변환을 준비한 뒤 Data Lineage에서 스케줄을 생성하고 Parameterization을 켭니다. 대상 데이터셋의 변환에 선언된 매개변수가 표시됩니다.

Standard 모드

Standard 모드는 스케줄에 값을 저장하고 실행할 때 변환으로 전달합니다. 기본값을 사용하려는 매개변수는 스케줄에서 숨깁니다.

예를 들어 town_name을 New York으로 바꾸고 risk_factor를 숨기면 지역은 New York으로 선택하고 risk factor는 변환의 기본값 5를 사용합니다.

시간별 증분 처리와 일별 전체 정리

한 변환을 매시간 증분 실행하되 하루에 한 번 전체 데이터를 다시 써서 정리하는 구성을 만들 수 있습니다. Boolean 매개변수 is_compaction_job의 기본값을 False로 둡니다.

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


@incremental(v2_semantics=True)
@transform.using(
    processed_towns=Output("/examples/towns/incremental"),
    towns=Input("/examples/towns/input"),
    is_compaction_job=BooleanParam(False),
)
def compact_towns(processed_towns, towns, is_compaction_job):
    if is_compaction_job.value:
        towns_df = towns.polars(mode="current")
        processed_towns.set_mode("replace")
    else:
        towns_df = towns.polars(mode="added")
    processed_towns.write_table(towns_df)

정리 실행은 현재 입력 전체를 읽고 출력을 대체합니다. 일반 실행은 추가 입력을 읽습니다. 출력 쓰기 모드만 replace로 바꾸고 입력은 추가분만 읽는 구성과 구분해야 과거 데이터를 누락하지 않습니다. 전체 정리에는 전체 입력을 읽고 쓰는 자원과 시간이 필요합니다.

• 시간별 스케줄은 매개변수화를 생략하여 기본값 False를 사용합니다.
• 일별 스케줄은 매개변수화를 켜고 is_compaction_job=True를 전달합니다.

이 구성은 같은 변환을 서로 다른 목적으로 실행하려고 스케줄을 둘로 나누는 경우입니다. 일반적인 중복 스케줄과 구분하여 범위와 실행 주기를 관리합니다.

Parallelized 모드와 run setup

Parallelized 모드는 서로 다른 매개변수 값을 가진 독립 빌드를 병렬로 실행하고 결과를 합칩니다. 값은 스케줄에 직접 고정하지 않고 run setup이라는 실행 설정에 저장합니다.

run setup은 매개변수 값의 묶음입니다. 스케줄을 만들 때 매개변수 하나를 run setup key로 지정하고 각 설정에는 그 키의 고유한 값을 줍니다. 서로 다른 설정은 다른 실행 설정이 끝날 때까지 기다리지 않고 빌드할 수 있습니다.

기존 스케줄의 Metrics & History에서 Run setups 탭을 엽니다.

Union view 출력

병렬 실행의 결과는 스케줄 주 브랜치의 대상 데이터셋 내용을 갱신하는 방식으로 모이지 않습니다. 스케줄에 구성한 출력 union view에 성공한 실행 설정의 행들을 모읍니다.

스케줄을 생성하거나 편집할 때 각 대상 데이터셋에 결과를 모을 union view를 연결합니다. 예시에서는 processed_towns의 결과를 processed_towns_union으로 모읍니다.

union view의 기반 데이터와 현재 뷰의 의미는 View에서 설명합니다.

액션에서 실행

스케줄을 실행하는 액션은 전달받은 매개변수로 run setup을 만들고 바로 빌드를 시작합니다. 같은 run setup key가 이미 있으면 기존 설정의 값을 갱신한 다음 빌드합니다.

따라서 업무 애플리케이션이 서로 다른 지역이나 시나리오의 값을 전달하여 독립 결과를 만들고, union view에서 함께 소비하는 구성에 사용할 수 있습니다.

제한