Snowpark: Python DataFrames on Snowflake

Every other lesson in this track writes SQL directly. Snowpark is Snowflake’s Python API: you build a query as chained DataFrame calls, and on a live account it compiles to one SQL statement that runs on a warehouse — the same warehouse, the same billing, the same query_history from the architecture lesson. This lesson uses Snowflake’s own local testing framework instead of a live account, so every example below runs with no warehouse and no cost — and it draws a hard line around exactly what that mode can and cannot show you.

A local testing session

from snowflake.snowpark import Session

session = Session.builder.config("local_testing", True).create()
print(session)

$ python session_setup.py
<snowflake.snowpark.session.Session: account="MOCK_ACCOUNT", role="MOCK_ROLE", database="MOCK_DATABASE", schema="MOCK_SCHEMA", warehouse="MOCK_WAREHOUSE">

local_testing=True is a real, first-party Snowpark feature for unit-testing pipelines offline — not a workaround. Every identifier reports MOCK_* because there is no account behind it; the DataFrame API underneath is the real one.

DataFrames are lazy — a plan, not a result

orders = session.create_dataframe(
    [(1001, 1, 25.50, "completed"), (1002, 2, 12.00, "completed"),
     (1003, 1, 40.00, "returned"),  (1004, 3, 8.75, "completed"),
     (1005, 2, 63.20, "pending")],
    schema=["order_id", "customer_id", "amount", "status"],
)

completed = orders.filter(orders["status"] == "completed")
print(type(completed))

$ python lazy_plan.py
<class 'snowflake.snowpark.dataframe.DataFrame'>

filter() returned immediately with another DataFrame — no rows were touched. Nothing runs until you call .show() or .collect(), exactly like building a SQL query string before executing it.

completed.show()

-----------------------------------------------------
|"ORDER_ID"  |"CUSTOMER_ID"  |"AMOUNT"  |"STATUS"   |
-----------------------------------------------------
|1001        |1              |25.5      |completed  |
|1002        |2              |12.0      |completed  |
|1004        |3              |8.75      |completed  |
-----------------------------------------------------

Aggregating like SQL, chained like Python

from snowflake.snowpark.functions import col, sum as sql_sum, count

revenue_by_customer = (
    orders
    .filter(col("status") == "completed")
    .group_by("customer_id")
    .agg(sql_sum(col("amount")).alias("total"), count("*").alias("orders"))
    .sort("customer_id")
)
revenue_by_customer.show()

$ python aggregate.py
--------------------------------------
|"CUSTOMER_ID"  |"TOTAL"  |"ORDERS"  |
--------------------------------------
|1              |25.5     |1         |
|2              |12.0     |1         |
|3              |8.75     |1         |
--------------------------------------

Read this as the group by/sum/count SQL from the architecture lesson written as method calls instead of a string — on a live account, that is literally what it becomes.

Python UDFs run locally too

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

@udf(name="tier_for_amount", return_type=StringType(), replace=True)
def tier_for_amount(amount: float) -> str:
    if amount >= 40:
        return "gold"
    if amount >= 15:
        return "silver"
    return "bronze"

orders.select(col("order_id"), tier_for_amount(col("amount")).alias("tier")).show()

$ python udf.py
-----------------------
|"ORDER_ID"  |"TIER"  |
-----------------------
|1001        |silver  |
|1002        |bronze  |
|1003        |gold    |
|1004        |bronze  |
|1005        |gold    |
-----------------------

@udf registers a real Python function as a callable inside DataFrame expressions. On a live account this ships the function to Snowflake and runs it server-side, next to the data — local testing runs it in-process instead, which is why the logic is trustworthy to unit-test this way even though the execution location isn’t real.

Two real limits of local testing

Predict what happens if you drop to raw SQL:

try:
    session.sql("select 1").collect()
except NotImplementedError as e:
    print(f"{type(e).__name__}: {e}")

$ python raw_sql.py
NotImplementedError: [Local Testing] Session.sql is not supported.

Local testing only emulates the DataFrame API — not a SQL engine. Any pipeline that drops to session.sql() for something the DataFrame API doesn’t cover needs a live account to test.

The second limit is subtler. Ask a DataFrame for the SQL it would run:

print(completed.queries)

$ python inspect_queries.py
{'queries': ['SELECT MOCK_TEST_FAKE_QUERY()'], 'post_actions': []}

On a live account, .queries returns the actual compiled SQL — genuinely useful for understanding what Snowflake will execute and for EXPLAIN-style review before running something expensive. Local testing can’t show it, because there’s no real compiler behind this session: the placeholder is Snowpark telling you, honestly, “this part isn’t real.” Trust local testing for pipeline logic; verify the generated SQL and its cost against a real warehouse before it goes anywhere near production.

Practice

orders.filter(col("status") == "pending").show()

----------------------------------------------------
|"ORDER_ID"  |"CUSTOMER_ID"  |"AMOUNT"  |"STATUS"  |
----------------------------------------------------
|1005        |2              |63.2      |pending   |
----------------------------------------------------

from snowflake.snowpark.functions import avg

orders.group_by("status").agg(avg(col("amount")).alias("avg_amount")).sort("status").show()

----------------------------------
|"STATUS"   |"AVG_AMOUNT"        |
----------------------------------
|completed  |15.416666666666666  |
|pending    |63.2                |
|returned   |40.0                |
----------------------------------

pdf = revenue_by_customer.to_pandas()
print(type(pdf))
print(pdf)

<class 'pandas.DataFrame'>
   CUSTOMER_ID  TOTAL  ORDERS
0            1  25.50       1
1            2  12.00       1
2            3   8.75       1

print(tier_for_amount(50))

Traceback (most recent call last):
  ...
TypeError: The input of UDF "MOCK_DATABASE"."MOCK_SCHEMA"."TIER_FOR_AMOUNT" must be Column, column name, or a list of them

Next: query profiling and cost control — finding the query that is spending your budget.