Snowflake에서 Snowpark 변환 실행하기

Python 저장소에 최신 transforms-tables를 설치하고 가상 테이블 공통 설정을 적용합니다. 선언한 Snowflake 가상 입력과 출력의 연결 정보로 Snowpark 세션이 만들어집니다. 모든 테이블은 같은 Snowflake 소스 연결로 접근할 수 있어야 하며 @incremental은 지원하지 않습니다.

Snowpark DataFrame과 UDF

SnowflakeInput.dataframe()은 Snowpark DataFrame을 반환하고 SnowflakeOutput.write()는 그 결과를 출력 테이블에 씁니다. DataFrame의 .session에서 같은 세션을 얻어 UDF를 등록할 수 있습니다. 함수에 타입 힌트가 없으면 input_types와 return_type으로 입력과 반환 자료형을 지정합니다.

다음 예제는 정수 ID를 CUSTOMER-NO- 접두사가 있는 문자열로 바꿉니다. NULL ID는 NULL로 유지합니다. 경로, RID와 외부 테이블 이름은 가상의 값입니다.

from snowflake.snowpark.functions import col, udf
from snowflake.snowpark.types import LongType, StringType
from transforms.api import transform
from transforms.tables import (
    SnowflakeInput,
    SnowflakeOutput,
    SnowflakeTable,
    TableInput,
    TableOutput,
)

SOURCE_RID = "ri.magritte..source.00000000-0000-0000-0000-000000000001"
ID_PREFIX = "CUSTOMER-NO-"


@transform.snowflake.using(
    source=TableInput("/Example Project/Inputs/customers"),
    output=TableOutput(
        "/Example Project/Outputs/customers_cleaned", SOURCE_RID,
        SnowflakeTable("DATABASE", "PUBLIC", "CUSTOMERS_CLEANED"),
    ),
)
def compute(source: SnowflakeInput, output: SnowflakeOutput):
    rows = source.dataframe()
    session = rows.session

    @udf(session=session, input_types=[LongType()], return_type=StringType())
    def fix_id_col(ident):
        return None if ident is None else ID_PREFIX + str(ident)

    output.write(rows.with_column("ID", fix_id_col(col("ID"))))

UDF와 DataFrame 계산은 Snowflake에서 실행합니다. Foundry의 일반 Spark 선언 대신 transform.snowflake.using()을 사용합니다.

Warehouse 선택

기본 계산 자원은 소스에 구성한 warehouse입니다. 다른 warehouse를 사용하려면 .with_warehouse()로 지정합니다. 해당 연결이 지정한 warehouse를 사용할 수 있어야 합니다.

from transforms.api import transform
from transforms.tables import SnowflakeTable, TableInput, TableOutput

SOURCE_RID = "ri.magritte..source.00000000-0000-0000-0000-000000000001"


@transform.snowflake.using(
    source=TableInput("/Example Project/Inputs/snowflake_table"),
    output=TableOutput(
        "/Example Project/Outputs/warehouse_copy", SOURCE_RID,
        SnowflakeTable("DATABASE", "PUBLIC", "WAREHOUSE_COPY"),
    ),
).with_warehouse(warehouse="ANALYTICS_WAREHOUSE")
def compute(source, output):
    output.write(source.dataframe())

PySpark API를 사용하는 Snowpark Connect

.with_engine(engine="pyspark")를 연결하면 Snowpark Connect 세션을 사용합니다. 함수의 입력/출력 타입도 SnowparkConnectInput과 SnowparkConnectOutput으로 구분합니다. 입력의 .dataframe()은 Spark Connect DataFrame이고 .session은 Spark Connect SparkSession입니다.

from pyspark.sql.connect.functions import lit
from transforms.api import transform
from transforms.tables import (
    SnowflakeTable,
    SnowparkConnectInput,
    SnowparkConnectOutput,
    TableInput,
    TableOutput,
)

SOURCE_RID = "ri.magritte..source.00000000-0000-0000-0000-000000000001"


@transform.snowflake.using(
    source=TableInput("/Example Project/Inputs/customers"),
    output=TableOutput(
        "/Example Project/Outputs/snowpark_connect_result", SOURCE_RID,
        SnowflakeTable("DATABASE", "PUBLIC", "CUSTOMERS_CONNECT"),
    ),
).with_engine(engine="pyspark")
def compute(source: SnowparkConnectInput, output: SnowparkConnectOutput):
    rows = source.dataframe().withColumn("test", lit(3))
    output.write_dataframe(rows)

계산은 Snowflake에 연결된 엔진에서 실행되며, 일반 Foundry Spark cluster를 생성하는 설정은 아닙니다. 사용할 연산은 Snowpark Connect가 제공하는 PySpark API 범위 안에 있어야 합니다.

작은 결과를 Pandas로 수집하여 처리하기

Snowpark의 to_pandas()는 쿼리를 실행하고 결과 전체를 Foundry의 로컬 Python 메모리로 가져옵니다. Pandas가 환경에 설치되어 있어야 하며 데이터가 해당 메모리에 들어가는 경우에만 사용합니다. 이 단계에서는 표 계산 전체가 외부에 머무르는 것은 아닙니다.

다음 예제는 CITY, STATE, ZIP_CODE 문자열을 순서대로 이어 붙여 SHA-256 해시를 계산하고 ID, ANON_CODE만 다시 씁니다. 우편번호의 앞자리 0을 보존하려면 ZIP_CODE를 문자열로 유지합니다. 열 이름 ANON_CODE와 별개로, 결정적 해시만으로 개인 정보의 익명성이 보장되지는 않습니다. 값의 범위가 좁으면 사전 대입으로 재식별할 수 있습니다.

import hashlib

import pandas as pd
from transforms.api import transform
from transforms.tables import SnowflakeTable, TableInput, TableOutput

SOURCE_RID = "ri.magritte..source.00000000-0000-0000-0000-000000000001"


@transform.snowflake.using(
    source=TableInput("/Example Project/Inputs/customers"),
    output=TableOutput(
        "/Example Project/Outputs/customer_codes", SOURCE_RID,
        SnowflakeTable("DATABASE", "PUBLIC", "CUSTOMERS_CLEANED_ANON"),
    ),
)
def compute(source, output):
    remote = source.dataframe()
    local = remote.to_pandas()
    addresses = local[["CITY", "STATE", "ZIP_CODE"]].itertuples(index=False, name=None)
    local["ANON_CODE"] = pd.Series(
        [
            hashlib.sha256(f"{city}{state}{zipcode}".encode("utf-8")).hexdigest()
            for city, state, zipcode in addresses
        ],
        index=local.index,
        dtype="string",
    )
    result = local[["ID", "ANON_CODE"]]
    output.write(remote.session.create_dataframe(result))

이 해시의 입력 정의는 구분자 없는 문자열 연결입니다. 서로 다른 필드 조합이 같은 연결 문자열을 만들 수 있으며, NULL이나 숫자 등 다른 값 유형의 표현 규칙도 업무 계약에 맞게 정해야 합니다. 필드 경계를 보존하는 직렬화로 바꾸면 기존 해시와 값이 달라지므로 같은 식별자 체계로 취급하지 않습니다.

Pandas DataFrame을 session.create_dataframe()에 넘기면 Snowflake 세션의 임시 테이블에 데이터가 저장되고 이를 참조하는 DataFrame을 얻습니다. 임시 테이블은 세션 종료 시 정리됩니다. 로컬 변환 후에는 외부로 다시 업로드하는 비용과 권한도 필요합니다.

Pandas 수집 기능이 입력/출력의 동일 소스 가상 테이블 조건을 해제하지는 않습니다. Foundry 데이터셋과 외부 표를 한 변환에서 조합하려면 Foundry-native Spark 경로나 별도 변환을 사용합니다.