SnowPro Specialty: Snowpark Complete Guide — Python/Scala Data Processing

Snowpark is a development framework for processing data inside Snowflake programmatically from Python, Scala, and Java. Every DataFrame operation runs on a Snowflake warehouse, so your data never has to leave Snowflake. Lazy evaluation means no SQL is issued until you call an action, and the Snowflake Optimizer generates the best execution plan for you.

SnowPro Specialty: Snowpark Exam Overview

Exam Domains and Weights

Core DataFrame API Methods

Snowpark's DataFrame API lets you write data processing as Spark-style method chains. Every method compiles down to SQL, so the Snowflake Optimizer handles the optimization for you.

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

session = Session.builder.configs(connection_params).create()

# Lazy evaluation: no SQL is executed at this point
df = session.table("SALES")
result = (
    df.filter(col("SALE_DATE") >= "2026-01-01")
      .group_by("REGION", "PRODUCT_CATEGORY")
      .agg(
          sum_("AMOUNT").alias("TOTAL_SALES"),
          avg("AMOUNT").alias("AVG_SALE"),
          count("ORDER_ID").alias("ORDER_COUNT"),
      )
      .sort(col("TOTAL_SALES").desc())
)

# collect() issues the SQL and runs it on Snowflake
rows = result.collect()
for row in rows:
    print(f"{row['REGION']}: {row['TOTAL_SALES']}")

UDF / UDTF / Stored Procedure Comparison

UDF Definition Example (Python)

from snowflake.snowpark.functions import udf
from snowflake.snowpark.types import StringType, IntegerType

# Inline UDF
@udf(name="calculate_tax", return_type=IntegerType(),
     input_types=[IntegerType()], replace=True, is_permanent=False)
def calculate_tax(amount: int) -> int:
    return int(amount * 0.1)

# SQL call: SELECT calculate_tax(AMOUNT) FROM ORDERS;

# --- UDTF definition example ---
from snowflake.snowpark.functions import udtf
from snowflake.snowpark.types import StructType, StructField

class SplitTags:
    def process(self, tags: str):
        for tag in tags.split(","):
            yield (tag.strip(),)

session.udtf.register(
    SplitTags,
    output_schema=StructType([StructField("TAG", StringType())]),
    input_types=[StringType()],
    name="split_tags",
    replace=True,
)
# SQL call: SELECT * FROM TABLE(split_tags('ml,ai,data'));

Stored Procedure Definition Example

from snowflake.snowpark.functions import sproc

@sproc(name="refresh_summary", replace=True, is_permanent=True,
       stage_location="@DEPLOY_STAGE",
       packages=["snowflake-snowpark-python"])
def refresh_summary(session: Session) -> str:
    source = session.table("RAW_EVENTS")
    summary = (
        source.group_by("EVENT_TYPE")
              .agg(count("*").alias("CNT"))
    )
    summary.write.save_as_table("EVENT_SUMMARY", mode="overwrite")
    return "EVENT_SUMMARY refreshed"

# Execution: CALL refresh_summary();

Vectorized UDF (Pandas UDF)

A Vectorized UDF takes pandas.Series as input and output, processing in batches instead of row by row, so it runs much faster than a regular UDF. It is well-suited to performance-sensitive numerical computation and ML inference.

from snowflake.snowpark.functions import pandas_udf
from snowflake.snowpark.types import PandasSeriesType, IntegerType
import pandas as pd

@pandas_udf(name="batch_normalize", return_type=PandasSeriesType(IntegerType()),
            input_types=[PandasSeriesType(IntegerType())],
            replace=True)
def batch_normalize(series: pd.Series) -> pd.Series:
    mean = series.mean()
    std = series.std()
    return ((series - mean) / std * 100).astype(int)

# SQL: SELECT batch_normalize(SCORE) FROM EXAM_RESULTS;
# Batch processing is dozens of times faster than a row-wise UDF

Snowpark ML

Snowpark ML is a library that lets you train and serve ML models entirely inside Snowflake. It exposes a scikit-learn-compatible API for fit/predict, and models are registered to the Snowflake Model Registry for production deployment.

from snowflake.ml.modeling.linear_model import LogisticRegression
from snowflake.ml.registry import Registry

# Training
train_df = session.table("TRAINING_DATA")
model = LogisticRegression(
    input_cols=["FEATURE_A", "FEATURE_B", "FEATURE_C"],
    label_cols=["LABEL"],
    output_cols=["PREDICTION"],
)
model.fit(train_df)

# Inference
test_df = session.table("TEST_DATA")
predictions = model.predict(test_df)
predictions.write.save_as_table("PREDICTIONS", mode="overwrite")

# Model registration
reg = Registry(session)
mv = reg.log_model(
    model,
    model_name="churn_classifier",
    version_name="v1",
    sample_input_data=train_df.limit(10),
)

Snowpark Container Services (SPCS)

SPCS is a managed service for running Docker containers inside Snowflake. While Snowpark focuses on DataFrame operations and UDF/Sproc execution, SPCS runs general-purpose workloads such as GPU-based model training, custom REST APIs, and full-stack web applications.

Study Roadmap

Sample Question

Frequently Asked Questions