단일 노드, Spark와 Databricks 구문

현재 Python transforms는 실행 방식을 decorator에서 명시합니다. transform.using은 단일 노드가 기본이고 Spark는 transform.spark.using을 사용합니다. 예전 @transform(...)를 새 .using(...)으로 단순 치환하면 실행 방식도 달라질 수 있으므로 사용하는 DataFrame API를 함께 맞춰야 합니다.

단일 노드의 권장 구문

transforms 3.68.0 이상과 transformsLangPythonPluginVersion 1.978.0 이상에서 사용할 수 있습니다.

from transforms.api import Input, Output, transform


@transform.using(
    output=Output("/Example Project/Outputs/adults"),
    source=Input("/Example Project/Inputs/people"),
)
def clean(output, source):
    rows = source.pandas()
    output.write_table(rows.loc[rows["age"] >= 18])

단일 노드를 명시적으로 나타내는 @transform.lightweight(...)도 사용할 수 있습니다. 위의 @transform.using(...) 대신 같은 입출력 바인딩을 이 decorator에 전달합니다.

계속 지원하는 Legacy 구문

기존 @lightweight와 @transform(...) 조합도 전체 지원합니다. 새 API를 쓰지 못하는 환경에서 사용할 수 있습니다.

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


@lightweight
@transform(
    output=Output("/Example Project/Outputs/adults"),
    source=Input("/Example Project/Inputs/people"),
)
def clean(output, source):
    rows = source.pandas()
    output.write_table(rows.loc[rows["age"] >= 18])

두 코드는 같은 출력에 대한 대안입니다. 한 Pipeline에 둘을 동시에 등록하지 않습니다. 입력에는 age 열이 있고 환경에 Pandas가 있어야 합니다.

Spark 구문

transforms 3.95.0 이상과 plugin 1.1003.0 이상에서는 다음을 권장합니다.

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


@transform.spark.using(
    output=Output("/Example Project/Outputs/spark_adults"),
    source=Input("/Example Project/Inputs/people"),
)
def clean(output, source):
    rows = source.dataframe().filter(F.col("age") >= 18)
    output.write_dataframe(rows)

단일 노드의 Pandas/Polars와 Spark DataFrame 및 쓰기 메서드를 서로 바꾸어 쓰지 않습니다.

Databricks Connect

Databricks로 계산을 내려보내는 변환은 @transform.databricks.using(...)을 사용합니다. 이전에 따로 조합하던 Databricks 및 transform decorator를 통합한 구문입니다.

Databricks 소스와 입력 가상 테이블을 먼저 준비해야 합니다. TableOutput은 Foundry에 등록할 출력 경로, Databricks source RID와 Unity Catalog의 대상 이름을 받습니다. 출력은 외부 시스템에 기록됩니다.

기본 세션은 serverless compute를 사용합니다. 특정 클러스터를 선택하려면 .with_compute(cluster_id=...)를 연결합니다. 다음 상수들은 형식을 보여 주는 예시이며 실제 프로젝트 경로, source RID, 클러스터 ID와 대상 이름으로 바꿔야 합니다.

from pyspark.sql.functions import col
from transforms.api import transform
from transforms.tables import DatabricksInput, DatabricksOutput, TableInput, TableOutput

SOURCE_RID = "ri.magritte..source.11111111-1111-1111-1111-111111111111"
CLUSTER_ID = "example-cluster-id"
TARGET_TABLE = "CATALOG.SCHEMA.TABLE"


@transform.databricks.using(
    source_table=TableInput("/Example Project/Tables/input"),
    output_table=TableOutput(
        "/Example Project/Tables/output",
        SOURCE_RID,
        TARGET_TABLE,
    ),
).with_compute(cluster_id=CLUSTER_ID)
def compute(source_table: DatabricksInput, output_table: DatabricksOutput):
    rows = source_table.dataframe().filter(col("id") > 1)
    output_table.write_dataframe(rows)

입력에는 id 열이 있어야 합니다. serverless를 사용할 경우 특정 cluster_id 지정을 하지 않습니다.

원격 Python 의존성

Databricks의 Python UDF가 패키지를 요구하면 .with_dependencies(dependencies=[...])로 설치할 조건을 지정합니다. 아래는 실제 pyjokes 라이브러리로 문자열 열을 생성하는 예입니다.

모듈 수준 import를 사용하는 이 코드에는 저장소의 작성 환경에도 pyjokes<1이 있어야 합니다. Libraries에서 추가하고, 원격 Databricks 실행 환경에는 with_dependencies로 같은 조건을 전달합니다.

import pyjokes
from pyspark.sql.functions import udf
from pyspark.sql.types import StringType
from transforms.api import transform
from transforms.tables import DatabricksOutput, TableOutput

SOURCE_RID = "ri.magritte..source.11111111-1111-1111-1111-111111111111"


@udf(returnType=StringType())
def joke_for_row():
    return pyjokes.get_joke()


@transform.databricks.using(
    output_table=TableOutput(
        "/Example Project/Tables/jokes",
        SOURCE_RID,
        "CATALOG.SCHEMA.JOKES",
    ),
).with_dependencies(dependencies=["pyjokes<1"])
def compute(output_table: DatabricksOutput):
    rows = output_table.spark_session.range(1, 10)
    output_table.write_dataframe(rows.withColumn("jokes", joke_for_row()))

이 예시는 별도 cluster 설정이 없어 기본 serverless 세션을 사용합니다. 실제 소스와 외부 대상에 대한 권한이 필요하며, 라이브러리 목록이 외부 데이터 접근 권한을 대신하지 않습니다.

이전 버전의 오류

Usage of transform.using() on outdated repository 또는 A function object does not have an attribute using이면 라이브러리와 템플릿 버전을 확인합니다. 저장소 업그레이드를 적용하거나 지원되는 Legacy 단일 노드 구문을 선택합니다.

버전 숫자만 수동으로 바꾸기보다 실제 템플릿과 실행 라이브러리의 버전 연결을 함께 맞춥니다.

Keyboard shortcuts