Snowpark (Python DataFrame, Snowpark Connect, 컨테이너 런타임)

Snowpark 이 하는 일

Snowpark 은 Python, Java, Scala 로 쓴 DataFrame 코드를 SQL 로 번역해 Snowflake 웨어하우스에서 실행하는 라이브러리입니다. 데이터는 Snowflake 밖으로 나오지 않고, collect() 나 to_pandas() 를 부를 때만 결과가 클라이언트로 옵니다.

• DataFrame 연산은 지연 실행입니다. filter, select, join 을 이어 붙이는 동안에는 아무 것도 실행되지 않고, collect, show, save_as_table 같은 동작에서 한 번에 SQL 이 만들어집니다.
• Python UDF, 저장 프로시저, UDTF 도 Snowpark 으로 등록합니다. 이것들은 웨어하우스 안의 보안 샌드박스 Python 에서 돕니다. 자세한 것은 UDF 와 저장 프로시저 에 있습니다.

세션 연결

from snowflake.snowpark import Session
from snowflake.snowpark.functions import col, sum as sum_, date_trunc

session = Session.builder.configs({
    "account": "myorg-prod",
    "user": "SVC_ETL",
    "authenticator": "SNOWFLAKE_JWT",          # 키 페어 인증
    "private_key_file": "/secrets/rsa_key.p8",
    "role": "ETL_ROLE",
    "warehouse": "ETL_WH",
    "database": "ANALYTICS",
    "schema": "MART",
}).create()

주의 2026년 8월부터 10월 사이 계정별로 적용되는 강한 인증 의무화로 사람 사용자는 비밀번호만으로, 서비스 사용자는 비밀번호로 로그인할 수 없습니다. 스크립트와 파이프라인은 키 페어, OAuth, 프로그래밍 방식 액세스 토큰(PAT), Workload Identity Federation 중 하나로 옮겨야 합니다. 자세한 일정은 인증 에 있습니다.

기본 패턴

orders = session.table("RAW.ORDERS")

daily = (
    orders.filter(col("STATUS") == "PAID")
          .with_column("D", date_trunc("day", col("ORDERED_AT")))
          .group_by("D", "COUNTRY")
          .agg(sum_("AMOUNT").alias("REVENUE"))
)

daily.write.mode("overwrite").save_as_table("MART.DAILY_REVENUE")

# 실행될 SQL 확인
print(daily.queries["queries"][-1])
daily.explain()

pandas 를 쓰고 싶다면

snowflake.snowpark.modin.plugin 을 쓰면 pandas 문법으로 작성한 코드를 Snowflake 안에서 실행할 수 있습니다 (pandas on Snowflake). 익숙한 문법을 유지하면서 데이터를 내려받지 않는다는 점이 장점이지만, 모든 pandas API 가 같은 속도로 번역되지는 않으므로 큰 작업은 실행 계획과 시간을 꼭 확인합니다.

Snowpark Connect for Apache Spark

Snowpark Connect 는 Spark Connect 프로토콜로 기존 PySpark DataFrame 과 Spark SQL 코드를 거의 그대로 Snowflake 에서 실행하게 해 줍니다. 2025년 11월 BUILD 에서 GA 로 발표되었고, 2026년 9월에는 Spark 작업을 보는 Spark Monitoring UI 가 Preview 로 나왔습니다.

• 클러스터를 따로 운영하지 않고 Snowflake 웨어하우스가 실행합니다.
• RDD API, 일부 Spark 설정, 사용자 정의 Scala 코드처럼 지원하지 않는 부분이 있으므로 이전 전에 호환성 검사를 돌립니다.
• 마이그레이션 도구 묶음은 2026 Summit 에서 Snowflake AIM 으로 합쳐졌습니다 (SnowConvert AI 와 Snowpark Migration Accelerator 포함).

웨어하우스 고르기

Snowpark-optimized 웨어하우스는 RESOURCE_CONSTRAINT 로 메모리 배수(MEMORY_1X, MEMORY_16X, MEMORY_64X)와 CPU 아키텍처를 고르고, 같은 크기의 표준 웨어하우스보다 시간당 크레딧이 큽니다. 예를 들어 MEMORY_16X 의 M 은 시간당 6 크레딧입니다 (Service Consumption Table 2026-10-15 시행판).

CREATE WAREHOUSE ds_wh
  WAREHOUSE_TYPE = 'SNOWPARK-OPTIMIZED'
  WAREHOUSE_SIZE = 'MEDIUM'
  RESOURCE_CONSTRAINT = 'MEMORY_16X';

패키지 관리

• 웨어하우스 런타임의 Python 패키지는 Snowflake Anaconda 채널에서 고릅니다. 사용 전에 조직 관리자가 Anaconda 약관을 승인해야 합니다.
• 채널에 없는 순수 Python 패키지는 스테이지에 올려 imports 로 씁니다.
• 2026년 5월에 계정 기본 패키지 저장소를 정하는 DEFAULT_PYTHON_ARTIFACT_REPOSITORY 매개변수가 추가되었습니다.
• 컨테이너 런타임(Notebooks, ML Jobs)은 PyPI 패키지를 쓸 수 있지만 외부 접근 통합(External Access Integration)이 필요합니다.

성능 함정

1. 루프 안의 collect(): 반복마다 쿼리가 하나씩 나가 클라우드 서비스와 웨어하우스 시간이 쌓입니다. 한 번의 DataFrame 연산으로 바꿉니다.
2. to_pandas() 로 수억 행 내려받기: 네트워크와 클라이언트 메모리가 병목입니다.
3. 캐시 없이 같은 DataFrame 재사용: 여러 번 쓰는 중간 결과는 cache_result() 로 임시 테이블에 담습니다.
4. 일반 UDF 로 대량 처리: 벡터화 UDF(@pandas_udf 또는 vectorized)가 수 배 빠릅니다.

트러블슈팅

• 증상: 로컬에서는 잘 되던 코드가 저장 프로시저로 올리면 "ModuleNotFoundError" 를 냅니다.
• 원인: 프로시저의 packages 나 imports 에 패키지를 등록하지 않았습니다.
• 해결: session.sproc.register(..., packages=["pandas", "scikit-learn"]) 처럼 명시하고, 채널에 없는 패키지는 스테이지에 올려 imports 로 넣습니다.

공식 문서

• Snowpark API
• Snowpark Connect for Spark
• Snowpark-optimized warehouses
• pandas on Snowflake