Snowflake + Python

You are a Snowflake data engineer using Snowpark for Python. The user wants to build and execute Python-based data transformations and ML pipelines directly within Snowflake using Snowpark.

What to check first

• Run pip list | grep snowpark to verify Snowpark for Python is installed (version 1.0+)
• Confirm Snowflake account credentials (account identifier, user, password, warehouse, database, schema)
• Check that your Snowflake account has compute resources (warehouse) active and accessible

Steps

1. Install Snowpark: pip install snowpark (includes pandas, numpy dependencies)
2. Create a Snowflake Session object using Session.builder() with account, user, password, warehouse, database, and schema parameters
3. Load data into a DataFrame using session.table() for existing tables or session.create_dataframe() for Python objects
4. Apply transformations using Snowpark DataFrame API methods like .select(), .filter(), .with_column(), and .group_by()
5. Use snowpark.functions for SQL-equivalent operations (e.g., col(), when(), lag(), dense_rank())
6. For ML: use snowpark.ml.preprocessing and snowpark.ml.modeling to apply sklearn-compatible models at scale
7. Execute .collect() to materialize results locally or .write.mode().save_as_table() to persist in Snowflake
8. Close the session with session.close() when done

Code

from snowpark.session import Session
from snowpark.functions import col, when, avg, count
from snowpark.types import StructType, StructField, StringType, IntegerType, DoubleType
import pandas as pd

# 1. Initialize Snowflake Session
connection_params = {
    "account": "xy12345.us-east-1",
    "user": "your_username",
    "password": "your_password",
    "warehouse": "compute_wh",
    "database": "analytics_db",
    "schema": "public"
}

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

# 2. Load data from Snowflake table
df = session.table("sales_data")

# 3. Transform using Snowpark DataFrame API
df_transformed = (
    df
    .filter(col("amount") > 100)
    .with_column("revenue_category", 
                 when(col("amount") > 1000, "high")
                 .when(col("amount") > 500, "medium")
                 .otherwise("low"))
    .group_by("product_id", "revenue_category")
    .agg(
        count("*").alias("transaction_count"),
        avg("amount").alias("avg_amount")
    )
)

# 4. Create a Snowpark DataFrame from Python data
local_data

Note: this example was truncated in the source. See the GitHub repo for the latest full version.

Common Pitfalls

• Treating this skill as a one-shot solution — most workflows need iteration and verification
• Skipping the verification steps — you don't know it worked until you measure
• Applying this skill without understanding the underlying problem — read the related docs first

When NOT to Use This Skill

• When a simpler manual approach would take less than 10 minutes
• On critical production systems without testing in staging first
• When you don't have permission or authorization to make these changes

How to Verify It Worked

• Run the verification steps documented above
• Compare the output against your expected baseline
• Check logs for any warnings or errors — silent failures are the worst kind

Production Considerations

• Test in staging before deploying to production
• Have a rollback plan — every change should be reversible
• Monitor the affected systems for at least 24 hours after the change