transform: Spark와 경량 변환 선언

transforms.api.transform은 compute 함수를 변환 객체로 감쌉니다. 선언에서 붙인 입력, 출력 및 다른 매개변수의 이름은 함수 인자 이름과 같아야 합니다. 실행 시 각 명세의 유형에 맞는 객체를 생성하여 인자로 주입합니다. Spark 계산의 한 단계를 나타내는 Transform은 호출 가능한 객체이며 일반적으로 이 데코레이터로 구성합니다.

Spark 변환

transforms.api.transform(**kwargs)

kwargs는 이름이 있는 Param 또는 그 하위 클래스의 명세입니다. compute 함수가 직접 출력을 기록해야 합니다. 두 가상 입력을 각각 별도 출력에 기록하는 예시입니다.

from transforms.api import Input, Output, transform


@transform(
    first_input=Input("/path/to/first/input/dataset"),
    second_input=Input("/path/to/second/input/dataset"),
    first_output=Output("/path/to/first/output/dataset"),
    second_output=Output("/path/to/second/output/dataset"),
)
def my_compute_function(first_input, second_input, first_output, second_output):
    first_output.write_dataframe(first_input.dataframe())
    second_output.write_dataframe(second_input.dataframe())

입출력 객체와 반환 DataFrame의 기록 책임을 구분합니다. transform_df는 반환값을 단일 출력에 자동으로 기록하지만 transform은 함수에서 각 출력에 쓰기 메서드를 호출합니다.

선택적 ctx 인자

compute 함수에 ctx 인자를 추가하면 실행 맥락을 받습니다. ctx를 입출력 명세 이름으로 사용하지 않습니다. 일반 Spark 실행은 TransformContext, 증분 Spark 실행은 IncrementalTransformContext, 경량 실행은 LightweightContext를 사용합니다.

Spark 맥락에는 abort_job, auth_header, environment, fallback_branches, parameters, spark_session이 있습니다. 다음은 실행 매개변수의 abort 값이 참이면 작업을 중단하는 패턴입니다.

from transforms.api import Input, Output, transform


@transform(
    my_input=Input("/path/to/input"),
    my_output=Output("/path/to/output"),
)
def my_compute_function(ctx, my_input, my_output):
    if ctx.parameters.get("abort"):
        ctx.abort_job()
    my_output.write_dataframe(my_input.dataframe())

경량 맥락은 abort_job, auth_header, is_incremental을 제공합니다. 비증분 실행을 거부하려는 정책은 다음처럼 표현할 수 있습니다. 이 예시는 초기 비증분 실행도 중단하므로 새 파이프라인을 초기화하는 기본 변환으로 사용하지 않습니다.

from transforms.api import Input, Output, transform


@transform.using(
    my_input=Input("/path/to/input"),
    my_output=Output("/path/to/output"),
)
def reject_nonincremental_run(ctx, my_input, my_output):
    if not ctx.is_incremental:
        ctx.abort_job()
    my_output.write_pandas(my_input.pandas())

실제로 증분 실행을 구성하려면 다음과 같이 incremental을 적용합니다. 작업 중단은 모든 출력 트랜잭션을 중단하고 실행을 끝냅니다.

증분 실행

incremental은 출력 빌드 이력을 읽어 마지막 입력 상태를 파악합니다. Spark 입력, 출력 및 맥락을 증분 대응 유형으로 바꾸며 경량 ContainerTransform도 지원합니다.

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


@incremental(v2_semantics=True)
@transform(
    first_input=Input("/path/to/first/input/dataset"),
    first_output=Output("/path/to/first/output/dataset"),
)
def copy_added_rows(first_input, first_output):
    first_output.write_dataframe(first_input.dataframe())

경량 예시에서는 pandas를 run 의존성으로 추가합니다.

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


@incremental(v2_semantics=True)
@transform.using(
    my_input=Input("/path/to/input"),
    my_output=Output("/path/to/output"),
)
def copy_added_rows(my_input, my_output):
    my_output.write_pandas(my_input.pandas())

조인, 집계와 distinct 같은 연산은 이전 출력과 새 입력을 함께 고려하는 증분 논리가 필요합니다. 모드와 조건은 incremental을 확인합니다.

using과 자원 구성

transforms.api.transform.using(**kwargs)

using은 Spark 없는 단일 노드 경량 변환을 만드는 클래스 메서드입니다. pandas, Polars와 파일시스템 API 등을 사용할 수 있지만 일반 Spark API의 일부만 지원하며 Spark 프로파일을 사용하지 않습니다.

using에는 Input, Output 같은 입출력 명세를 이름별로 전달합니다. 값 매개변수도 입력 명세로 선언할 수 있습니다. 이 선언용 명세와 compute에 주입되는 LightweightInput 및 LightweightOutput의 역할을 구분합니다. 자원이나 컨테이너 옵션은 using의 입력 이름에 섞지 않고 반환된 빌더에 메서드를 연결하여 설정합니다.

기본 자원을 사용하는 예시:

from transforms.api import Input, Output, transform


@transform.using(
    my_input=Input("/path/to/input"),
    my_output=Output("/path/to/output"),
)
def compute_func(my_input, my_output):
    my_output.write_pandas(my_input.pandas())

CPU와 메모리를 지정하는 예시:

from transforms.api import Input, Output, transform


@transform.using(
    my_input=Input("/path/to/input"),
    my_output=Output("/path/to/output"),
).with_resources(cpu_cores=4, memory_gb=8)
def compute_func(my_input, my_output):
    my_output.write_pandas(my_input.pandas())

Polars DataFrame을 받아 반환하려면 transform_polars를 사용할 수 있으며 polars를 run 의존성으로 추가합니다. 일반 Spark의 원시 파일 접근은 FileSystem, 경량 실행은 FoundryDataSidecarFileSystem을 사용합니다.

컨테이너 설정

with_container로 이미지와 태그를 지정합니다. 다음 예시는 my-image:0.0.1이 코드 저장소의 backing Artifacts 저장소에 있으며, USER_WORKING_DIR 아래에 실행 가능한 실제 data_generator 프로그램이 준비된 상황입니다. 프로그램은 인자 없이 실행되어 현재 작업 디렉터리에 헤더가 있는 data.csv를 생성해야 합니다. pandas를 run 의존성으로 추가합니다.

프로그램이 실패하면 출력을 계속 기록하지 않도록 종료 코드를 확인합니다.

import os
import subprocess
from pathlib import Path

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


@transform.using(
    my_output=Output("/path/to/output")
).with_container(
    container_image="my-image",
    container_tag="0.0.1",
)
def run_data_generator_executable(my_output):
    working_dir = Path(os.environ["USER_WORKING_DIR"])
    executable = working_dir / "data_generator"
    subprocess.run([str(executable)], cwd=str(working_dir), check=True)
    my_output.write_table(pd.read_csv(working_dir / "data.csv"))

컨테이너 워크플로를 먼저 활성화해야 하며 BYOC는 Preview를 지원하지 않습니다. 이미지 조건과 기존 경량 데코레이터의 호환 경로는 lightweight를 확인합니다. 위 프로그램은 사용자가 준비하는 실제 외부 실행 파일이며, 변환 코드는 그 파일의 실행과 CSV 출력 수신을 담당합니다.