Python 변환에서 가상 테이블 사용하기

가상 테이블은 데이터를 Foundry에 저장하지 않은 채 지원되는 외부 데이터 플랫폼의 표를 조회하고 쓰는 자산입니다. Python 변환에서는 transforms-tables의 TableInput과 TableOutput으로 참조합니다.

두 계산 방식

Foundry-native 가상 테이블 입력을 일반 lightweight의 Pandas나 Polars로 직접 읽는 방식은 지원하지 않습니다. Compute pushdown은 lightweight에서 외부 계산을 조율하는 별도 경로입니다. Snowpark 결과를 작은 Pandas 표로 가져오는 기능은 결과의 명시적인 로컬 수집이며 일반 TableInput의 경량 엔진 지원과 구분합니다.

Compute pushdown에는 @incremental 증분 계산을 사용할 수 없습니다.

저장소와 소스 준비

1. Python 저장소를 최신 템플릿으로 업그레이드합니다.
2. Libraries에서 transforms-tables를 설치합니다.
3. 소스에서 Code imports를 활성화하여 코드 저장소가 소스를 가져와 사용할 수 있게 합니다.
4. 소스의 Export controls에서 이 변환의 입력에 허용할 보안 표지와 조직을 정합니다.
5. 출력에 사용할 소스를 저장소의 사이드바에서 가져옵니다.

가상 테이블을 사용하는 변환은 소스에 구성한 네트워크 egress 정책에 따라 외부 통신 권한을 얻습니다. use_external_systems decorator와 가상 테이블은 같은 변환에서 함께 사용할 수 없습니다. 소스 기반 외부 변환을 사용하거나 가상 테이블 처리와 해당 decorator를 사용하는 처리를 별도 변환으로 나눕니다.

가상 테이블에서 가상 테이블로 쓰기

다음은 Snowflake에 등록한 가상 입력을 Foundry Spark로 읽어 다른 가상 출력에 쓰는 예제입니다. 경로, 소스 RID와 외부 위치는 가상의 값이며 실제 소스와 테이블로 지정해야 합니다.

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

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


@transform.spark.using(
    source=TableInput("/Example Project/Inputs/external_table"),
    output=TableOutput(
        "/Example Project/Outputs/external_copy",
        SOURCE_RID,
        SnowflakeTable("DATABASE", "SCHEMA", "COPIED_TABLE"),
    ),
)
def compute(source, output):
    output.write_dataframe(source.dataframe())

Foundry-native Spark에서는 입력과 출력이 같은 소스나 같은 외부 플랫폼일 필요가 없습니다.

Foundry 데이터셋을 외부 테이블에 쓰기

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

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


@transform.spark.using(
    source=Input("/Example Project/Inputs/foundry_dataset"),
    output=TableOutput(
        "/Example Project/Outputs/exported_table",
        SOURCE_RID,
        SnowflakeTable("DATABASE", "SCHEMA", "EXPORTED_TABLE"),
    ),
)
def compute(source, output):
    output.write_dataframe(source.dataframe())

외부 저장 위치를 지정하는 Table API는 생성 이후 소스와 위치를 변경할 수 없는 조건도 설명합니다.

Snowflake로 계산 밀어내기

Snowflake 연결의 가상 테이블만 사용하면 Snowpark DataFrame 연산을 외부 엔진에 실행할 수 있습니다. 다음 예제는 NEW_COLUMN에 ABC를 넣습니다.

from snowflake.snowpark.functions import lit
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/snowflake_enriched",
        SOURCE_RID,
        SnowflakeTable("DATABASE", "SCHEMA", "ENRICHED_TABLE"),
    ),
)
def compute(source, output):
    output.write(source.dataframe().with_column("NEW_COLUMN", lit("ABC")))

외부 계산의 공통 조건은 compute pushdown을 따릅니다. 연결별 세션은 BigQuery, Databricks, Snowflake의 설정 방법에 맞춰 구성합니다.

JDBC 읽기 분할

JDBC로 가상 테이블을 읽을 때 .dataframe(options=...)에 분할 설정을 전달하여 병렬로 읽습니다. 분할 열은 숫자, 날짜 또는 timestamp여야 합니다.

from transforms.api import Output, transform
from transforms.tables import TableInput


@transform.spark.using(
    source=TableInput("/Example Project/Inputs/external_table"),
    output=Output("/Example Project/Outputs/partitioned_read"),
)
def compute(source, output):
    rows = source.dataframe(options={
        "numPartitions": "3",
        "partitionColumn": "column_name",
        "lowerBound": "1",
        "upperBound": "100",
    })
    output.write_dataframe(rows)

이 설정은 1과 100 사이에서 분할 간격을 계산하여 세 범위로 읽습니다. lowerBound와 upperBound는 행 필터가 아니므로 지정 범위 밖의 값도 반환합니다. 범위를 제한하려면 별도의 데이터 필터가 필요합니다.

파일 템플릿 마법사의 가상 출력

Code Repositories의 파일 템플릿은 virtual table 변수를 사용하여 입력과 출력을 구성할 수 있습니다. 가상 출력 생성 기능은 Beta이며 기능이 변경될 수 있습니다. 마법사에서 출력 소스와 Foundry 안의 가상 테이블 위치를 선택합니다.