transform_df: PySpark DataFrame 변환

transforms.api.transform_df는 PySpark DataFrame을 입력으로 받아 DataFrame을 반환하는 compute 함수에서 Transform을 구성합니다. 입력 이름은 함수 인자 이름과 일치해야 합니다. 출력은 이름 없는 위치 인자 Output 하나만 받으며 반환 DataFrame을 그 출력 데이터셋에 자동으로 기록합니다.

transforms.api.transform_df(output, **inputs)

두 입력의 열 순서와 자료형이 호환되는 가상 예시입니다. Spark union은 열 이름이 아닌 위치로 결합하므로 열 순서를 맞춥니다.

from transforms.api import Input, Output, transform_df


@transform_df(
    Output("/path/to/output/dataset"),
    first_input=Input("/path/to/first/input/dataset"),
    second_input=Input("/path/to/second/input/dataset"),
)
def my_compute_function(first_input, second_input):
    return first_input.union(second_input)

증분 실행

incremental로 감싸면 입력은 added 모드로 읽고 쓰기 모드는 incremental이 결정합니다. 기본값 이외의 모드가 필요하면 transform을 사용합니다.

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


@incremental(v2_semantics=True)
@transform_df(
    Output("/path/to/output/dataset"),
    first_input=Input("/path/to/first/input/dataset"),
)
def compute(first_input):
    return first_input

위 증분 예시는 입력의 추가 행을 그대로 추가 출력으로 전달하는 경우입니다. 조인이나 집계의 과거 상태 처리는 별도의 증분 논리로 구성합니다.