Snowflake 연결, 가상 테이블과 연산 위임

Snowflake 소스는 Snowflake 데이터를 Foundry에 동기화하거나 가상 테이블로 연결합니다. 원천 탐색, 대량 가져오기, 증분 수집, Iceberg 동기화, 가상 테이블, 연산 위임과 테이블 내보내기는 일반 제공(GA)입니다. 이전 Export tasks는 Sunset 단계입니다.

데이터 이동 방향을 구분해야 합니다. Snowflake 소스의 Iceberg 동기화는 원천 데이터를 Foundry 관리형 Iceberg 테이블에 씁니다. Horizon Catalog의 Iceberg 가상 테이블은 외부에 있는 테이블을 복사하지 않고 등록합니다. 반대로 Snowflake에서 Foundry 관리형 Iceberg 테이블을 읽고 쓰는 구성은 Foundry Iceberg REST 카탈로그와 Snowflake를 연결하는 별도 경로입니다.

연결 속성

Data Connection의 +New Source에서 Snowflake 소스를 선택합니다. Account identifier에 밑줄이 있으면 대시로 바꿔야 합니다. 예를 들어 my_account_prod는 my-account-prod로 입력하며, 밑줄을 그대로 두면 네트워크 문제가 발생합니다.

인증 방식

Snowflake는 비밀번호를 사용하는 사람 사용자에게 MFA를 요구하고 서비스 사용자의 비밀번호 사용을 막는 변경을 진행하고 있습니다. 비밀번호 인증에 계속 의존하는 운영 연결은 다른 방식으로 전환해야 합니다.

모든 인증 방식에서 데이터베이스와 스키마의 사용 권한 및 대상 테이블의 조회 권한이 필요합니다. 가상 테이블에서는 설정한 warehouse의 사용 권한도 필요합니다.

기본 네트워크 허용 목록

Foundry 워커 소스에는 실제 Snowflake 계정의 호스트와 포트에 대한 외부 통신 정책을 추가합니다. Snowflake 콘솔에서 다음 질의를 실행하여 대상을 확인할 수 있습니다.

SELECT t.VALUE:type::VARCHAR as type,
       t.VALUE:host::VARCHAR as host,
       t.VALUE:port as port
FROM TABLE(FLATTEN(input => PARSE_JSON(SYSTEM$ALLOWLIST()))) AS t;

최소한 SNOWFLAKE_DEPLOYMENT와 STAGE 항목이 필요합니다. 일반적으로 환경의 기본 공용 게이트웨이 IP에서 연결하지만, AWS 간 또는 Azure 간 같은 클라우드 내부 트래픽은 다른 경로를 사용할 수 있으며 PrivateLink 구성이 필요할 수 있습니다.

AWS의 내부 S3 stage

Snowflake가 내부 S3 stage를 사용자 정의 VPCE로 라우팅하면 JDBC 드라이버가 Foundry VPC에서 도달할 수 없는 주소로 접속할 수 있습니다. 해당 주소 형식은 다음과 같습니다.

<bucketname>.bucket.vpce-<vpceid>.s3.<region>.vpce.amazonaws.com

이 경우 Connection details의 JDBC 속성에 S3_STAGE_VPCE_DNS_NAME을 추가하고 값을 비워 둡니다. 빈 값은 null 설정에 해당하며 문자열 "null"을 넣는 뜻이 아닙니다. 그러면 S3 stage 트래픽은 다음과 같은 S3 Gateway Endpoint 경로를 사용하여 공용 인터넷을 경유하지 않는 연결을 유지합니다.

<bucketname>.bucket.s3.<region>.vpce.amazonaws.com

Foundry와 같은 리전의 S3 버킷에 의존하는 외부 통신 정책은 해당 버킷의 정책도 Foundry 연결 요구에 맞게 구성해야 합니다.

Azure의 내부 stage

Azure에 있는 Snowflake는 데이터를 읽을 때 내부 stage 저장소에 직접 연결할 수 있습니다. Azure 기반 Foundry는 Azure 서비스 엔드포인트를 사용하므로 stage 저장소 접근을 별도로 허용해야 합니다.

먼저 SYSTEM$ALLOWLIST()의 전체 도메인 목록과 Azure 저장소 계정 리소스 식별자를 준비합니다. Standard 또는 Enterprise Edition이면 저장소 계정 식별자를 Snowflake 지원에 요청합니다.

Business Critical Edition에서는 계정의 ENABLE_INTERNAL_STAGES_PRIVATELINK를 TRUE로 설정한 뒤 SYSTEM$GET_PRIVATELINK_CONFIG()를 호출할 수 있습니다. 응답의 privatelink-internal-stage가 Azure 저장소 계정 리소스 식별자를 제공합니다. 실제 통신을 PrivateLink로 하지 않더라도 이 식별자가 필요합니다.

/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Storage/storageAccounts/{storageAccountName}

이후 다음 두 정책을 구성합니다.

1. 내부 stage 도메인을 포함해 SYSTEM$ALLOWLIST()가 반환하는 모든 대상의 표준 외부 통신 정책을 소스에 연결합니다.
2. 저장소 계정 리소스 식별자를 사용하여 Azure storage policy를 만듭니다.

소스 탐색과 동기화가 가능한지 확인하여 연결 구성을 완료합니다.

Snowflake 가상 테이블

Foundry 워커가 필요하며 에이전트 워커는 지원하지 않습니다. 연결의 warehouse를 지정하고 그 사용 권한을 부여합니다. Code Repositories에서 사용할 경우 가상 테이블과 소스의 코드 사용 설정도 수행합니다.

증분 읽기

테이블에 Change Tracking과 적절한 보존 기간의 Time Travel을 활성화합니다. 증분 기능은 CHANGES에 의존하며 버전이 없는 뷰에서는 사용할 수 없습니다.

Python Transforms의 current, added 읽기 모드를 지원합니다. METADATA$ACTION을 기준으로 관련 변경 행을 제공하고 다음 열을 함께 사용할 수 있습니다.

METADATA$ACTION
METADATA$ISUPDATE
METADATA$ROW_ID

접근 권한

원하는 상속 범위에 따라 데이터베이스, 스키마 또는 테이블에 권한을 적용합니다.

설정한 warehouse의 사용 권한도 필요합니다. Iceberg 테이블에는 테이블이 저장된 external volume의 USAGE가 추가로 필요합니다.

Horizon Catalog의 Iceberg 테이블

다음 조건을 갖추면 커넥터가 Iceberg 기능을 자동으로 노출합니다.

• Foundry에서 테이블의 external volume 저장소에 접근할 외부 통신 정책을 구성합니다.
• 소스의 인증 정보에 Horizon Catalog에서 발급하는 저장소 접근 인증 정보를 받을 권한이 있습니다.

Foundry는 Horizon Catalog의 Iceberg REST API와 인증 정보 발급 기능을 사용하여 기반 객체 저장소를 직접 읽고 씁니다. 데이터 읽기와 쓰기에 Snowflake 연산을 사용하지 않지만, 테이블 유형 판별 같은 일부 메타데이터 질의에는 소스의 warehouse를 여전히 사용합니다.

Horizon Catalog에 쓰는 기능은 Snowflake의 비공개 미리보기(Private preview) 단계입니다. 두 조건 중 하나라도 충족하지 못하면 일반 Snowflake 테이블처럼 연결하고 읽기와 쓰기에 warehouse를 사용합니다.

external volume마다 외부 통신 정책을 만들고 저장소 측의 네트워크 통제도 Foundry 트래픽을 허용하도록 구성해야 합니다. 같은 클라우드나 리전의 저장소라면 앞서 설명한 별도 네트워크 조건도 적용됩니다.

Iceberg 동기화와 연산 위임

Iceberg 동기화는 Foundry 워커에서 비증분 또는 증분 배치로 Snowflake 데이터를 Foundry 관리형 Iceberg 테이블에 씁니다. 여러 쓰기 모드를 지원합니다. 외부 테이블을 그대로 등록하는 Horizon 가상 테이블과 구분합니다.

일반 Snowflake 가상 테이블 입력은 Snowflake Spark 커넥터의 조건절 위임을 활용합니다. 같은 Snowflake 소스에 등록한 가상 테이블을 파이프라인의 입력과 출력으로 사용하면 연산 전체를 Snowflake로 위임하는 구성을 사용할 수 있습니다. Python Transforms 또는 Pipeline Builder의 외부 파이프라인에서 설정합니다.

OIDC 소스를 코드에서 사용

Python 저장소에 snowflake-connector-python과 예제의 polars 의존성을 추가하고 OIDC 소스를 가져옵니다. 코드의 소스 및 데이터셋 RID, 계정, warehouse와 데이터베이스를 실제 값으로 바꿉니다. my_table, col1, col2는 예시 테이블과 열입니다.

Snowflake는 이 경로에서 Foundry가 발급한 짧은 수명의 토큰을 OAuth 토큰으로 사용합니다. get_session_credentials().get().access_token으로 현재 토큰을 가져오며 별도의 정적 Snowflake 비밀번호를 코드에 넣지 않습니다.

읽기

최대 100행을 질의하고 열 이름과 행을 Polars DataFrame으로 구성해 출력합니다. 커서와 연결은 finally에서 닫습니다.

import polars as pl
from transforms.api import transform, Output, LightweightOutput
from transforms.external.systems import ResolvedSource, external_systems, Source
import snowflake.connector


@external_systems(
    snowflake_source=Source("<source_rid>")
)
@transform.using(
    output=Output("<output_dataset_rid>")
)
def read_from_snowflake(output: LightweightOutput, snowflake_source: ResolvedSource):
    conn = snowflake.connector.connect(
        authenticator="oauth",
        token=snowflake_source.get_session_credentials().get().access_token,
        account="<account_identifier>",
        warehouse="<warehouse_name>",
        database="<database_name>",
    )

    cursor = conn.cursor()
    try:
        cursor.execute("SELECT * FROM my_table LIMIT 100")
        columns = [col[0] for col in cursor.description]
        rows = cursor.fetchall()
    finally:
        cursor.close()
        conn.close()

    df = pl.DataFrame(rows, schema=columns)
    output.write_table(df)

쓰기

입력의 col1, col2를 대상 my_table에 삽입합니다. 테이블을 자동 생성하거나 기존 키를 갱신하는 코드는 아니며, 대상 테이블과 필요한 쓰기 권한을 미리 준비해야 합니다. 실행하면 외부 Snowflake 데이터가 변경됩니다.

import polars as pl
from transforms.api import transform, Input, Output, LightweightOutput
from transforms.external.systems import ResolvedSource, external_systems, Source
import snowflake.connector


@external_systems(
    snowflake_source=Source("<source_rid>")
)
@transform.using(
    output=Output("<output_dataset_rid>"),
    source_data=Input("<input_dataset_rid>"),
)
def write_to_snowflake(output: LightweightOutput, snowflake_source: ResolvedSource, source_data):
    conn = snowflake.connector.connect(
        authenticator="oauth",
        token=snowflake_source.get_session_credentials().get().access_token,
        account="<account_identifier>",
        warehouse="<warehouse_name>",
        database="<database_name>",
    )

    cursor = conn.cursor()
    try:
        df = source_data.dataframe()
        for row in df.iter_rows(named=True):
            cursor.execute(
                "INSERT INTO my_table (col1, col2) VALUES (%s, %s)",
                (row["col1"], row["col2"]),
            )
    finally:
        cursor.close()
        conn.close()

    output.write_table(df)

반정형 자료형

Snowflake의 array, object, variant는 값마다 구조가 달라질 수 있어 Foundry에서는 string으로 해석합니다. 예를 들어 Snowflake 배열 [ 1, 2, 3 ]은 Foundry 문자열 "[1,2,3]"으로 들어옵니다. 후속 변환에서 구조를 해석할 때 원천 자료형과 수집 자료형을 구분합니다.