transform_pandas: pandas DataFrame 변환

transforms.api.transform_pandas는 pandas DataFrame을 받고 반환하는 compute 함수를 변환으로 등록합니다. 기본 Spark 경로에서는 계산 전에 PySpark DataFrame을 pandas DataFrame으로 바꾸고 계산 후 다시 변환합니다. 출력은 위치 인자 Output 하나이며 입력은 이름별 Input 명세입니다.

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

저장소의 meta.yml에 pandas를 run 의존성으로 추가합니다. Polars DataFrame을 사용하려면 transform_polars를 선택합니다.

두 입력 결합

가상 데이터셋 경로를 사용하는 예시입니다. pandas의 concat 함수로 행을 결합합니다.

import pandas as pd
from transforms.api import Input, Output, transform_pandas


@transform_pandas(
    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 pd.concat([first_input, second_input], axis=0)

증분 실행

incremental로 감쌀 수 있습니다. 이 경로의 읽기 모드는 added이고 쓰기 모드는 incremental이 결정합니다. 다른 모드가 필요하면 transform을 사용합니다.

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


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

메모리와 사전 필터링

pandas 변환은 메모리에 들어가는 크기의 데이터셋에만 사용합니다. Spark의 toPandas는 선택한 데이터를 드라이버 메모리로 가져옵니다. 큰 입력을 먼저 줄여야 하면 transform_df에서 Spark 필터를 적용한 뒤 pandas로 변환하고, 처리 결과를 SparkSession.createDataFrame으로 돌려줍니다.

다음 가상 예시는 문자열 열 county와 category를 선택하고, county 값이 UK인 행의 누락된 category를 채웁니다. 빈 결과에서도 스키마를 추론하려 하지 않도록 선택한 Spark 스키마를 명시합니다.

from transforms.api import Input, Output, transform_df


@transform_df(
    Output("/path/to/output/dataset"),
    first_input=Input("/path/to/first/input/dataset"),
)
def prepare_subset(ctx, first_input):
    selected = first_input.filter(first_input.county == "UK").select(
        "county", "category"
    )
    pandas_frame = selected.toPandas()
    pandas_frame["category"] = pandas_frame["category"].fillna("unknown")
    return ctx.spark_session.createDataFrame(pandas_frame, schema=selected.schema)

기본 Spark 실행 맥락은 TransformContext입니다. 기존 lightweight 데코레이터와 조합한 단일 노드 경로는 Spark 변환의 기본 동작과 구분합니다.