PowderInsights | From PySpark to Snowpark: Migrating Python Data Engineering into Snowflake

Most Snowflake teams we work with still have a Spark cluster somewhere. It usually exists for one reason: at some point someone needed Python for a transformation that SQL made awkward — a fuzzy match, a scoring model, a nested JSON flattening job — and Spark was the tool at hand. Years later that cluster is a second platform to patch, secure, size and pay for, and the data it processes lives in Snowflake anyway.

Snowpark for Python removes most of the reasons to keep it. This tutorial covers what a realistic PySpark-to-Snowpark migration looks like: what maps one-to-one, what does not, and where teams lose weeks if nobody warns them.

What Snowpark actually is

Snowpark is a client-side DataFrame API that compiles to SQL and executes in a Snowflake virtual warehouse, plus a server-side Python runtime for UDFs, UDTFs and stored procedures. Two things follow from that:

DataFrame operations are lazy and become SQL. df.filter(...).group_by(...).agg(...) is not moving rows to your laptop; it builds a query plan that Snowflake executes. Nothing runs until you call an action such as collect(), show(), count() or save_as_table().
Python code in UDFs runs inside Snowflake, in a sandboxed interpreter with packages resolved from the Snowflake Anaconda channel (or from your own wheels on a stage).

That is why a Snowpark migration is mostly a code migration, not an infrastructure one. There is no cluster to size, no executor memory to tune, no shuffle partitions to guess at.

Connecting

from snowflake.snowpark import Session

session = Session.builder.configs({
    "account": "acme-prod",
    "user": "SVC_ETL",
    "authenticator": "SNOWFLAKE_JWT",
    "private_key_file": "/secrets/svc_etl.p8",
    "role": "ETL_ENGINEER",
    "warehouse": "WH_ETL_M",
    "database": "ANALYTICS",
    "schema": "STAGING",
}).create()

Use key-pair or OAuth, never a password in a config file — Snowflake now blocks single-factor password authentication for human users, and service accounts should have been on key-pair long before that.

Inside a Snowflake stored procedure or notebook you do not build a session at all; one is handed to you.

The translation table

The DataFrame API was deliberately modelled on Spark's, so most of a migration is mechanical.

PySpark	Snowpark for Python
`spark.read.parquet(path)`	`session.read.parquet("@stage/path")`
`spark.table("db.sch.t")`	`session.table("db.sch.t")`
`df.select(col("a"))`	`df.select(col("A"))`
`df.withColumn("x", ...)`	`df.with_column("X", ...)`
`df.groupBy(...).agg(...)`	`df.group_by(...).agg(...)`
`df.join(other, on=..., how="left")`	`df.join(other, on=..., how="left")`
`F.when(...).otherwise(...)`	`F.when(...).otherwise(...)`
`df.write.mode("overwrite").saveAsTable(t)`	`df.write.mode("overwrite").save_as_table(t)`
`df.repartition(200)`	(no equivalent — Snowflake handles parallelism)
`df.cache()` / `persist()`	`df.cache_result()` (materialises to a temp table)
`spark.sql("...")`	`session.sql("...")`

Two gotchas bite everyone:

Identifier casing. Snowflake upper-cases unquoted identifiers. col("customer_id") resolves to CUSTOMER_ID; if the column was created as "customer_id" with quotes, you must quote it in Snowpark too. Normalising casing during ingestion is worth the one-off pain.
collect() is a client operation. In Spark people call it carelessly on small results; in Snowpark it drags the whole result set over the wire into your Python process. Keep it for genuinely small outputs.

Snowpark pandas: the shortest path for pandas code

If the job you are migrating is pandas rather than Spark, you often do not need to rewrite it at all. Snowpark pandas (the Modin-backed API) executes the pandas interface against Snowflake:

import modin.pandas as pd
import snowflake.snowpark.modin.plugin  # registers the Snowflake engine

orders = pd.read_snowflake("ANALYTICS.RAW.ORDERS")

daily = (
    orders[orders["STATUS"] != "CANCELLED"]
    .assign(ORDER_DAY=orders["ORDER_TS"].dt.date)
    .groupby(["ORDER_DAY", "REGION"], as_index=False)["AMOUNT"]
    .sum()
)

daily.to_snowflake("ANALYTICS.MARTS.DAILY_SALES", if_exists="replace", index=False)

Same code shape, no df.to_pandas() in the middle, and the data never leaves Snowflake. Coverage of the pandas surface is broad but not total — anything unsupported falls back to a slower path, so profile the job after conversion rather than assuming.

A useful rule: Snowpark pandas for exploratory and medium-size dataframes; the Snowpark DataFrame API for production pipelines, because the DataFrame API gives you a predictable query plan you can read in the Query Profile.

Python that has to stay Python: UDFs and UDTFs

When a transformation genuinely needs Python, push it into the warehouse instead of pulling rows out.

from snowflake.snowpark.functions import udf
from snowflake.snowpark.types import StringType
import re

@udf(name="ANALYTICS.UTIL.NORMALISE_SKU",
     is_permanent=True,
     stage_location="@ANALYTICS.UTIL.PY_UDFS",
     packages=["snowflake-snowpark-python"],
     replace=True,
     return_type=StringType())
def normalise_sku(raw: str) -> str:
    if raw is None:
        return None
    return re.sub(r"[^A-Z0-9]", "", raw.upper())

Row-at-a-time UDFs are fine for cheap logic. For anything with per-call overhead — model inference, heavy library initialisation — use a vectorised (pandas) UDF, which hands your function batches as a pandas Series:

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

@pandas_udf(name="ANALYTICS.ML.SCORE_TEXT", is_permanent=True,
            stage_location="@ANALYTICS.UTIL.PY_UDFS",
            packages=["pandas", "scikit-learn"], replace=True,
            return_type=PandasSeriesType(FloatType()),
            input_types=[PandasSeriesType(StringType())])
def score_text(s: pd.Series) -> pd.Series:
    from joblib import load
    import sys, os
    model = load(os.path.join(sys._xoptions["snowflake_import_directory"], "model.joblib"))
    return pd.Series(model.predict_proba(s.fillna(""))[:, 1])

Use a UDTF when one input row produces many output rows (JSON explosion, sessionisation), and a stored procedure when you want the whole orchestration — several DataFrame steps, a MERGE, some logging — to run server-side on a schedule.

Orchestrating without a scheduler you have to run

CREATE OR REPLACE PROCEDURE ANALYTICS.STAGING.LOAD_DAILY_SALES()
  RETURNS STRING
  LANGUAGE PYTHON
  RUNTIME_VERSION = '3.11'
  PACKAGES = ('snowflake-snowpark-python')
  HANDLER = 'run'
AS
$$
def run(session):
    orders = session.table("RAW.ORDERS").filter("STATUS <> 'CANCELLED'")
    ...
    orders.write.mode("overwrite").save_as_table("MARTS.DAILY_SALES")
    return "ok"
$$;

CREATE OR REPLACE TASK ANALYTICS.STAGING.T_DAILY_SALES
  WAREHOUSE = WH_ETL_M
  SCHEDULE = 'USING CRON 30 5 * * * UTC'
AS CALL ANALYTICS.STAGING.LOAD_DAILY_SALES();

For pure SQL-shaped transformations, prefer dynamic tables over hand-rolled procedures — but a Snowpark procedure plus a task is the right answer when the logic is genuinely imperative.

Sizing and cost: unlearning Spark habits

Spark instincts translate badly.

Do not repartition, do not tune shuffles. There is no equivalent knob, and there does not need to be.
Warehouse size affects a single query's speed, not throughput per credit, for well-parallelised work. Start at S or M and only go up if the Query Profile shows spilling to local or remote storage.
Set AUTO_SUSPEND = 60 on ETL warehouses. Snowpark sessions that idle while your Python does client-side work will otherwise burn credits.
Watch for accidental round-trips. to_pandas() followed by more pandas work, then writing back, is the classic anti-pattern: it moves the whole dataset out and back for no reason.
Check the Query Profile after migration. Snowpark generates SQL; occasionally a chain of with_column calls produces a deeply nested query that is better rewritten as one select.

Testing

Snowpark code is ordinary Python, so pytest works. Two patterns carry most of the load: build small dataframes with session.create_dataframe([...], schema=[...]) and assert on collect(), and run the suite against an isolated database created by zero-copy clone in CI, dropped afterwards. Snowflake's local testing framework covers a useful subset of the API without a connection, which is handy for unit-testing transformation functions, but integration tests should hit a real account.

A migration order that works

Inventory the Spark jobs and split them: pure SQL logic, SQL-plus-a-little-Python, genuinely Python-heavy.
Kill the first bucket. Most "Spark ETL" is a SELECT with joins and aggregates; rewrite it as a dynamic table or a dbt model and delete the job entirely.
Convert the second bucket to the Snowpark DataFrame API with UDFs for the awkward bits.
Leave the third bucket last and consider Snowpark Container Services for anything needing a GPU, a long-running process, or a library the Anaconda channel does not carry.
Run both in parallel for one cycle, hash-compare outputs, then decommission the cluster — and remember to remove the egress paths, IAM roles and spend that came with it.

The payoff is not just credits. It is one security model, one set of governance policies, one place to look when a number is wrong.

Where teams get stuck

The two failures we are called in to fix: pipelines that were translated literally, keeping collect()-then-loop patterns that made sense on a cluster and now push millions of rows through a client process; and dependency drift, where a UDF pins a package version the Anaconda channel later moves past and the job breaks silently at the next deployment. Both are cheap to avoid at design time and expensive to unpick later.

PowderInsights builds and migrates Snowflake data platforms end to end, including Spark and pandas decommissioning, Snowpark pipeline design, UDF and container workloads, and the CI/CD around them. Get in touch with the jobs you are trying to retire and we will tell you which bucket each one falls into and what the conversion realistically costs.