lightweight: 기존 경량 변환 데코레이터

transforms.api.lightweight는 변환을 Spark 없는 단일 노드 실행으로 전환하는 기존 데코레이터입니다. 3.85.0부터 사용 중단 권고 상태이며 새 선언에는 transform.using을 사용합니다. 기존 데코레이터를 쓰려면 foundry-transforms-lib-python을 의존성으로 추가해야 합니다.

경량 변환은 작은 규모부터 중간 규모 데이터셋에서 실행 시간과 비용을 줄일 수 있습니다. pandas, Polars와 파일시스템 API 등 일반 변환 API의 일부를 지원합니다. pandas 또는 polars를 사용할 때는 해당 패키지를 meta.yml의 run 의존성으로 추가합니다. Spark 프로파일은 지원하지 않습니다.

transforms.api.lightweight(
    _maybe_function=None, *, cpu_cores=2, memory_mb=None, memory_gb=None,
    gpu_type=None, container_image=None, container_tag=None,
    container_shell_command=None,
)

memory_gb와 memory_mb 중 하나만 지정할 수 있습니다. container_image, container_tag, container_shell_command 중 하나라도 지정하면 이미지와 태그는 둘 다 지정해야 합니다.

셸 명령을 생략하면 Python 환경을 준비하고 데코레이터가 감싼 사용자 코드를 실행하는 기본 entrypoint를 사용합니다. 그 기본값은 데코레이터가 적용된 변환의 default_container_entrypoint 속성으로 확인합니다.

실행 맥락

compute 함수에 ctx를 추가하면 LightweightContext가 주입됩니다. 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 compute(ctx, my_input, my_output):
    if not ctx.is_incremental:
        ctx.abort_job()
    my_output.write_pandas(my_input.pandas())

기존 호출 형태

기존 코드의 괄호 없는 데코레이터:

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


@lightweight
@transform(
    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())

기본 목록에 나타나는 원시 파일을 바이트 단위로 복사하는 예시입니다. 기본 파일 목록에서는 숨김 파일을 제외합니다.

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


@lightweight()
@transform(
    my_input=Input("/path/to/input"),
    my_output=Output("/path/to/output"),
)
def copy_files(my_input, my_output):
    input_fs = my_input.filesystem()
    output_fs = my_output.filesystem()
    for file in input_fs.ls():
        with input_fs.open(file.path, "rb") as source:
            with output_fs.open(file.path, "wb") as target:
                while True:
                    chunk = source.read(1024 * 1024)
                    if not chunk:
                        break
                    target.write(chunk)

pandas DataFrame 데코레이터에 자원을 지정한 기존 경량 실행 예시:

from transforms.api import Input, Output, lightweight, transform_pandas


@lightweight(cpu_cores=8, memory_gb=3.5, gpu_type="NVIDIA_T4")
@transform_pandas(
    Output("/path/to/output"),
    my_input=Input("/path/to/input"),
)
def compute_func(my_input):
    return my_input

사용자 컨테이너 조건

container 계열 인자를 사용하는 방식은 BYOC입니다. 실행 시 코드 저장소의 파일을 USER_WORKING_DIR/user_code에서 사용할 수 있고 Python 환경도 제공됩니다. 이미지 자체에 Python을 미리 설치할 필요는 없습니다. 이미지는 코드 저장소의 backing Artifacts 저장소에서 사용할 수 있어야 합니다.

최소 이미지의 형식 예시:

FROM ubuntu:latest

RUN apt update && apt install -y coreutils curl sed

RUN useradd --uid 5001 user
USER 5001

IL5 또는 FedRAMP 환경에서 ubuntu:latest를 사용해 cURL 명령이 실패하면 ubuntu:jammy를 사용합니다. 컨테이너 워크플로 활성화, 이미지 게시와 backing 저장소 연결을 완료한 뒤 실행합니다. BYOC는 Preview를 지원하지 않으며 Build로 컨테이너를 실행합니다.

다음 기존 경량 선언은 실제 data_generator 실행 파일이 USER_WORKING_DIR 아래에 준비되어 있고, 작업 디렉터리에 헤더가 있는 data.csv를 생성한다는 조건을 사용합니다. pandas는 run 의존성으로 준비합니다.

import os
import subprocess
from pathlib import Path

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


@lightweight(container_image="my-image", container_tag="0.0.1")
@transform(my_output=Output("/path/to/output"))
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"))

실행 파일의 구현과 데이터 수집은 해당 프로그램의 책임이고, 이 변환은 종료 상태를 확인한 뒤 생성한 CSV를 기록합니다. 이미지가 최소 Dockerfile만으로 그 프로그램까지 제공한다고 가정하지 않습니다.