Python 변환 정의와 트랜잭션

변환은 decorator에 입출력 및 설정을 선언하고, 같은 이름의 함수 인자로 핸들을 받은 뒤 결과를 출력에 씁니다. 여러 정의를 Pipeline에 등록하여 데이터셋을 빌드합니다.

버전과 구문

새 .using을 지원하지 않는 환경에서는 Usage of transform.using() on outdated repository 또는 A function object does not have an attribute using 오류가 나타날 수 있습니다. 저장소와 라이브러리를 업그레이드하거나 지원되는 Legacy 구문을 사용합니다. 구문 이전은 각 실행 방식의 차이를 설명합니다.

단일 노드의 표 읽기

가상 입력에는 hair 열이 있고 Brown 값의 행을 선택한다고 가정합니다. 각 경로는 실제 자산으로 바꿉니다. 필요한 라이브러리는 환경에 설치해야 합니다.

Polars

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


@transform.using(
    hair_eye_color=Input("/Example Project/Inputs/students"),
    processed=Output("/Example Project/Outputs/polars_brown_hair"),
)
def filter_hair_color(hair_eye_color, processed):
    rows = hair_eye_color.polars()
    processed.write_table(rows.filter(pl.col("hair") == "Brown"))

DuckDB

ctx.duckdb().conn은 선언한 데이터셋 입력들을 함수의 입력 이름으로 등록합니다.

from transforms.api import Input, Output, transform


@transform.using(
    hair_eye_color=Input("/Example Project/Inputs/students"),
    processed=Output("/Example Project/Outputs/duckdb_brown_hair"),
)
def filter_hair_color(ctx, hair_eye_color, processed):
    query = ctx.duckdb().conn.sql(
        "SELECT * FROM hair_eye_color WHERE hair = 'Brown'"
    )
    processed.write_table(query)

Pandas

from transforms.api import Input, Output, transform


@transform.using(
    hair_eye_color=Input("/Example Project/Inputs/students"),
    processed=Output("/Example Project/Outputs/pandas_brown_hair"),
)
def filter_hair_color(hair_eye_color, processed):
    rows = hair_eye_color.pandas()
    processed.write_table(rows.loc[rows["hair"] == "Brown"])

읽을 수 있는 표 객체

write_table()에는 이 지원 객체들을 전달할 수 있습니다. 같은 데이터를 여러 형식으로 모두 읽어야 하는 것은 아닙니다. 실제 사용할 엔진의 객체만 읽어 불필요한 수집과 메모리 복사를 피합니다.

여러 입력과 출력

다음 단일 노드 예제는 hair_color에 id, hair, sex, eye_color에 id, eye가 있고 각 입력의 id가 고유하다는 계약입니다. 같은 id로 조인하여 갈색 머리 행을 성별 출력으로 나눕니다.

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


@transform.using(
    hair_color=Input("/Example Project/Inputs/hair_color"),
    eye_color=Input("/Example Project/Inputs/eye_color"),
    males=Output("/Example Project/Outputs/males"),
    females=Output("/Example Project/Outputs/females"),
)
def split_students(hair_color, eye_color, males, females):
    combined = hair_color.polars().join(eye_color.polars(), on="id", how="inner")
    selected = combined.filter(pl.col("hair") == "Brown")
    males.write_table(selected.filter(pl.col("sex") == "Male"))
    females.write_table(selected.filter(pl.col("sex") == "Female"))

Decorator의 이름과 함수 인자를 맞춰야 합니다. 두 출력을 같은 변환에 둔 실행 및 실패 결합은 다중 출력 설계를 따릅니다.

TransformContext

함수 인자 ctx는 런타임이 주입하는 TransformContext입니다. 따라서 입력이나 출력 자산 이름을 ctx로 선언하면 안 됩니다.

단일 노드에서는 위 DuckDB 연결처럼 엔진 문맥을 얻습니다. Spark에서는 ctx.spark_session으로 현재 session을 사용할 수 있습니다. 예를 들어 입력 없이 계산한 값을 출력하는 Spark 변환은 다음과 같습니다.

from pyspark.sql import functions as F
from transforms.api import Output, transform


@transform.spark.using(output=Output("/Example Project/Outputs/generated_squares"))
def generate_dataframe(ctx, output):
    rows = ctx.spark_session.range(3).withColumn("square", F.col("id") * F.col("id"))
    output.write_dataframe(rows)

Spark 변환

Spark 입력의 dataframe()은 PySpark DataFrame입니다. 일반적인 입력 및 출력 핸들 방식은 다음과 같습니다.

from pyspark.sql import functions as F
from transforms.api import Input, Output, transform


@transform.spark.using(
    hair_eye_color=Input("/Example Project/Inputs/students"),
    processed=Output("/Example Project/Outputs/spark_brown_hair"),
)
def filter_hair_color(hair_eye_color, processed):
    rows = hair_eye_color.dataframe()
    processed.write_dataframe(rows.filter(F.col("hair") == "Brown"))

같은 구조에서 여러 입력과 출력도 선언할 수 있습니다.

from pyspark.sql import functions as F
from transforms.api import Input, Output, transform


@transform.spark.using(
    hair_color=Input("/Example Project/Inputs/hair_color"),
    eye_color=Input("/Example Project/Inputs/eye_color"),
    males=Output("/Example Project/Outputs/spark_males"),
    females=Output("/Example Project/Outputs/spark_females"),
)
def split_students(hair_color, eye_color, males, females):
    combined = hair_color.dataframe().join(eye_color.dataframe(), on="id", how="inner")
    selected = combined.filter(F.col("hair") == "Brown")
    males.write_dataframe(selected.filter(F.col("sex") == "Male"))
    females.write_dataframe(selected.filter(F.col("sex") == "Female"))

출력 하나인 표 계산에는 transform_df가 입력 DataFrame을 바로 넘기고 반환값을 출력으로 씁니다.

from pyspark.sql import functions as F
from transforms.api import Input, Output, transform_df


@transform_df(
    Output("/Example Project/Outputs/returned_brown_hair"),
    students=Input("/Example Project/Inputs/students"),
)
def filter_students(students):
    return students.filter(F.col("hair") == "Brown")

단일 노드 자원은 with_resources, Spark 자원은 프로필로 구성합니다.

Iceberg 핸들

transforms.tables의 TableInput과 TableOutput으로 Iceberg 자산을 선언하면 계산 함수에 IcebergInput, IcebergOutput 같은 런타임 핸들을 받습니다. Decorator, 함수 인자와 계산의 세 부분은 같은 구조입니다. Iceberg Python 변환에서 읽기와 쓰기 계약을 적용합니다.

로직 버전과 staleness

Transform logic level versioning(TLLV)은 기본적으로 켜져 있습니다. 출력의 입력과 변환 버전이 모두 바뀌지 않았다면 최신으로 보고, 버전이 바뀌면 출력을 stale로 만들어 다시 계산합니다.

버전에는 변환을 정의한 모듈, 의존하는 모듈과 프로젝트 의존성이 포함됩니다. 별도 설정 파일처럼 이 범위 밖 파일의 변경도 반영하려면 Python 하위 프로젝트 build.gradle의 transformsPython 구성에서 tllvFiles 대상으로 지정합니다.

TLLV가 올바르게 동작하려면 import는 모듈 수준에 두고 다른 모듈의 객체를 patch하거나 변경하지 않아야 합니다. 이 조건을 만족하지 않는 프로젝트는 TLLV를 비활성화해야 합니다.

transformsPython {
    tllv false
}

출력 트랜잭션

변환은 출력에 연 트랜잭션 안에서 실행됩니다. 명시적으로 중단하지 않은 트랜잭션은 job이 성공하면 commit합니다. 선언한 출력에 실제 쓰기를 하지 않거나 쓰기 오류를 catch한 뒤 삼켜 성공시키면 해당 출력에 빈 스냅샷이 생길 수 있습니다.

출력별 쓰기 경로를 구현하고 실패는 실패로 전달해야 합니다. 기존 출력을 유지하는 조건에서는 작업 전체 중단을 사용합니다. 처리하지 않은 데이터를 정상 결과로 보이게 하는 성공 반환값이나 예외 무시는 트랜잭션의 의미를 바꿉니다.