Snowpark for Python: DataFrame API, Stored Procedures, and App Development

Snowflake is a powerful cloud data warehouse, but traditionally you interact with it using SQL. Snowpark for Python changes the game by allowing you to write data pipelines, stored procedures, and even full applications using Python, all while leveraging Snowflake's elastic compute and storage. This is a game-changer for data engineers and data scientists who prefer Python over SQL for complex transformations.

In this tutorial, you'll learn the core concepts of Snowpark: the DataFrame API, which provides a lazy evaluation model similar to PySpark but optimized for Snowflake; stored procedures that run Python code directly on Snowflake's virtual warehouses; and how to build end-to-end applications. We'll cover real-world scenarios like incremental data loads, custom business logic, and debugging in production. By the end, you'll be able to write production-ready Snowpark code that is efficient, maintainable, and scalable.

Whether you're migrating from PySpark or starting fresh, Snowpark offers a familiar API with the benefits of Snowflake's performance and security. Let's dive in.

Setting Up Snowpark for Python

Before you can use Snowpark, you need to install the library and establish a session. Snowpark is available as a Python package that you can install via pip. It requires Python 3.8 or later. You'll also need a Snowflake account and appropriate credentials.

First, install the Snowpark library:

``bash pip install snowflake-snowpark-python ``

Then, create a session by providing your Snowflake connection details. It's best practice to use a configuration file or environment variables rather than hardcoding credentials. Here's an example using a JSON config file:

```python from snowflake.snowpark import Session import json

with open('connection_params.json') as f: connection_params = json.load(f)

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

The connection_params.json file should contain:

``json { "account": "your_account", "user": "your_user", "password": "your_password", "role": "your_role", "warehouse": "your_warehouse", "database": "your_database", "schema": "your_schema" } ``

Once the session is created, you can start using the DataFrame API. Always close the session when done:

``python session.close() ``

For production, consider using a connection pool or Snowflake's key-pair authentication for better security.

from snowflake.snowpark import Session
import json

# Load connection parameters
with open('connection_params.json') as f:
    connection_params = json.load(f)

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

# Verify connection
print(session.sql('SELECT current_version()').collect())

# Close session
session.close()

The DataFrame API: Lazy Evaluation and Pushdown

Snowpark's DataFrame API is similar to PySpark but with a key difference: all operations are lazy and pushed down to Snowflake for execution. This means you can chain multiple transformations without triggering any computation until you call an action like collect(), show(), or to_pandas(). This pushdown optimization ensures that only the necessary data is transferred over the network.

Let's start by reading a table into a DataFrame:

``python df = session.table('sales_data') ``

You can also use SQL queries to create a DataFrame:

``python df = session.sql('SELECT * FROM sales_data WHERE year = 2023') ``

Now, let's perform some transformations. For example, filter, group by, and aggregate:

```python from snowflake.snowpark.functions import col, sum as sum_

result = (df .filter(col('amount') > 100) .group_by('region') .agg(sum_('amount').alias('total_sales')) .order_by('total_sales', ascending=False) )

result.show() ```

This will execute the SQL query on Snowflake and display the first 10 rows. You can also collect the results into a list of Row objects:

``python rows = result.collect() for row in rows: print(row.REGION, row.TOTAL_SALES) ``

One important aspect is that Snowpark DataFrames are immutable. Each transformation returns a new DataFrame. This allows for safe chaining and caching.

You can also join DataFrames:

``python df1 = session.table('customers') df2 = session.table('orders') joined = df1.join(df2, df1.customer_id == df2.customer_id, join_type='inner') ``

Snowpark supports all standard join types: inner, left, right, full, cross, and semi/anti joins.

For complex expressions, use the functions module which provides SQL-like functions such as when, cast, datediff, etc.

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

# Read table
df = session.table('sales_data')

# Transformations (lazy)
result = (df
    .filter(col('amount') > 100)
    .group_by('region')
    .agg(sum_('amount').alias('total_sales'))
    .order_by('total_sales', ascending=False)
)

# Action: execute and show
result.show()

# Collect all rows
rows = result.collect()
for row in rows:
    print(f"Region: {row.REGION}, Total Sales: {row.TOTAL_SALES}")

Writing Stored Procedures with Python

Snowpark allows you to create stored procedures using Python. These procedures run on Snowflake's compute resources and can be used to encapsulate business logic, perform complex transformations, or orchestrate multiple steps. Stored procedures are created using the session.sproc.register() decorator or by calling session.sproc.create_from_function().

Here's a simple example of a stored procedure that takes a date range and returns aggregated sales:

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

def aggregate_sales(session: Session, start_date: str, end_date: str) -> list: df = session.table('sales_data') result = (df .filter(col('sale_date').between(start_date, end_date)) .group_by('region') .agg(sum_('amount').alias('total_sales')) .collect() ) return [row.as_dict() for row in result]

# Register the stored procedure session.sproc.register( func=aggregate_sales, name='aggregate_sales_proc', return_type=[] # List of dicts )

# Call the stored procedure result = session.sql("CALL aggregate_sales_proc('2023-01-01', '2023-12-31')").collect() print(result) ```

Note that the first parameter of the function must be session: Session. The return type can be a Snowpark data type or a Python type. For complex returns, you can use Variant or Array.

Stored procedures can also be created with dependencies on other Python packages. You can specify a list of packages in the packages parameter:

``python session.sproc.register( func=my_proc, name='my_proc', packages=['pandas', 'numpy'] ) ``

This is useful for data science workflows where you need libraries like scikit-learn or pandas.

Stored procedures can also be written as SQL scripts that call Python functions, but the Python approach is more flexible.

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

def aggregate_sales(session: Session, start_date: str, end_date: str) -> list:
    df = session.table('sales_data')
    result = (df
        .filter(col('sale_date').between(start_date, end_date))
        .group_by('region')
        .agg(sum_('amount').alias('total_sales'))
        .collect()
    )
    return [row.as_dict() for row in result]

# Register
session.sproc.register(
    func=aggregate_sales,
    name='aggregate_sales_proc',
    return_type=[]
)

# Call
result = session.sql("CALL aggregate_sales_proc('2023-01-01', '2023-12-31')").collect()
print(result)

User-Defined Functions (UDFs) and Vectorized UDFs

Snowpark supports creating Python UDFs that can be used in SQL queries or DataFrame transformations. There are two types: scalar UDFs (one row in, one row out) and vectorized UDFs (batch processing). Vectorized UDFs are more efficient because they process batches of rows using pandas DataFrames.

Let's create a scalar UDF that calculates a discount:

```python from snowflake.snowpark.types import FloatType from snowflake.snowpark.functions import udf

def discount(price: float) -> float: return price * 0.9

# Register as UDF discount_udf = udf(discount, return_type=FloatType(), input_types=[FloatType()], name='discount_udf')

# Use in DataFrame df = session.table('products') df_with_discount = df.select(col('product_id'), col('price'), discount_udf(col('price')).alias('discounted_price')) df_with_discount.show() ```

For vectorized UDFs, you use the vectorized decorator:

```python from snowflake.snowpark.types import FloatType, PandasSeriesType from snowflake.snowpark.functions import pandas_udf

@pandas_udf(return_type=FloatType(), input_types=[PandasSeriesType(FloatType())]) def vectorized_discount(price_series: pd.Series) -> pd.Series: return price_series * 0.9

# Use in DataFrame df_with_discount = df.select(col('product_id'), col('price'), vectorized_discount(col('price')).alias('discounted_price')) ```

Vectorized UDFs are preferred for performance because they reduce Python function call overhead.

You can also create UDFs that return tables (UDTFs) using the udtf decorator.

from snowflake.snowpark.types import FloatType
from snowflake.snowpark.functions import udf, col

def discount(price: float) -> float:
    return price * 0.9

discount_udf = udf(discount, return_type=FloatType(), input_types=[FloatType()], name='discount_udf')

df = session.table('products')
df_with_discount = df.select(col('product_id'), col('price'), discount_udf(col('price')).alias('discounted_price'))
df_with_discount.show()

Building Applications with Snowpark

Snowpark is not just for ad-hoc queries; you can build full applications that run on Snowflake. For example, you can create a data pipeline that reads from a stage, transforms data, and writes to a table. Or you can build a REST API that uses Snowpark to serve data.

Snowpark provides a client-side API that runs in your application environment (e.g., a Python script or a web server). The DataFrame operations are pushed down to Snowflake, but the orchestration logic runs locally. This is ideal for ETL jobs.

Here's an example of an incremental load pipeline:

```python from snowflake.snowpark import Session from snowflake.snowpark.functions import col, max as max_

def incremental_load(session: Session, source_table: str, target_table: str, timestamp_col: str): # Get the last loaded timestamp from target last_load = session.table(target_table).select(max_(col(timestamp_col)).alias('max_ts')).collect()[0]['MAX_TS'] if last_load is None: last_load = '1900-01-01' # Read new data from source new_data = session.table(source_table).filter(col(timestamp_col) > last_load) # Write to target new_data.write.mode('append').save_as_table(target_table) print(f"Loaded {new_data.count()} rows")

# Run the pipeline incremental_load(session, 'raw_orders', 'orders', 'order_date') ```

For more complex applications, you can use Snowpark's DataFrame.save_as_table() with various modes: append, overwrite, errorifexists, etc.

You can also use Snowpark with orchestration tools like Airflow or Prefect. The Snowpark session can be created within a task and used to perform operations.

Another powerful feature is the ability to create Snowpark DataFrames from files in stages (e.g., Parquet, CSV). This allows you to process external data without loading it into a table first.

from snowflake.snowpark.functions import col, max as max_

def incremental_load(session, source_table, target_table, timestamp_col):
    last_load = session.table(target_table).select(max_(col(timestamp_col)).alias('max_ts')).collect()[0]['MAX_TS']
    if last_load is None:
        last_load = '1900-01-01'
    new_data = session.table(source_table).filter(col(timestamp_col) > last_load)
    new_data.write.mode('append').save_as_table(target_table)
    print(f"Loaded {new_data.count()} rows")

incremental_load(session, 'raw_orders', 'orders', 'order_date')

Optimization and Best Practices

To get the best performance from Snowpark, follow these best practices:

1. Use explain() to understand the query plan. This shows the SQL that will be executed. Look for full table scans, large shuffles, or inefficient joins.
2. Filter early and often. Push filters down to reduce the amount of data processed.
3. Use cache_result() for intermediate results that are reused. This materializes the DataFrame in Snowflake's cache, avoiding recomputation.
4. Avoid using collect() on large DataFrames. It pulls all data to the client. Use show() for sampling or to_pandas() with limits.
5. Use vectorized UDFs instead of scalar UDFs for better performance.
6. Choose the right warehouse size. Larger warehouses have more parallelism. For complex transformations, a larger warehouse can significantly reduce runtime.
7. Use partitioning and clustering keys on large tables to improve scan efficiency.
8. Monitor query performance using Snowflake's query history and the QUERY_HISTORY view.

Here's an example of using cache_result():

```python # Cache the filtered DataFrame filtered_df = df.filter(col('year') == 2023).cache_result()

# Use it multiple times without re-scanning agg1 = filtered_df.group_by('region').agg(sum_('sales')) agg2 = filtered_df.group_by('product').agg(avg_('price')) ```

Also, consider using Snowpark's DataFrame.union_all() and DataFrame.intersect() for set operations.

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

# Cache intermediate result
filtered_df = session.table('sales_data').filter(col('year') == 2023).cache_result()

# Use cached DataFrame
agg1 = filtered_df.group_by('region').agg(sum_('sales').alias('total_sales'))
agg2 = filtered_df.group_by('product').agg(avg('price').alias('avg_price'))

# Show results
agg1.show()
agg2.show()

Debugging and Logging in Snowpark

Debugging Snowpark code can be challenging because operations run remotely. Here are some techniques:

1. Use df.explain() to see the SQL. This helps you understand what will be executed.
2. Enable logging in stored procedures. You can use Python's logging module and write logs to a table or Snowflake's event table.
3. Use session.sql() to run diagnostic queries. For example, check row counts or sample data.
4. Use try-except blocks to catch exceptions and log them.
5. For UDFs, test with small data first. You can create a temporary table with a few rows and test your UDF.

Here's an example of logging in a stored procedure:

```python import logging

def my_proc(session: Session): logger = logging.getLogger('my_proc') logger.setLevel(logging.INFO) # Add a handler that writes to a table handler = logging.StreamHandler() logger.addHandler(handler) try: df = session.table('sales_data') logger.info(f"Processing {df.count()} rows") # ... transformations except Exception as e: logger.error(f"Error: {e}") raise ```

You can also use Snowflake's SYSTEM$LOG function to write logs to the event table.

For client-side debugging, use Python's debugger (pdb) or print statements. However, remember that DataFrame operations are lazy, so print statements inside transformations won't execute until an action is called.

import logging

def my_proc(session):
    logger = logging.getLogger('my_proc')
    logger.setLevel(logging.INFO)
    handler = logging.StreamHandler()
    logger.addHandler(handler)
    
    try:
        df = session.table('sales_data')
        logger.info(f"Processing {df.count()} rows")
        # Perform transformations
        result = df.filter(col('amount') > 100).group_by('region').count()
        result.show()
    except Exception as e:
        logger.error(f"Error: {e}")
        raise

# Call the procedure
my_proc(session)