Snowpark (with Python): A Beginner’s Guide

Photo by Darius Cotoi on Unsplash

Snowpark is a data engineering tool developed by Snowflake, that allows you to use your favorite programming languages (Python, Java, or Scala) to build and deploy data pipelines and analyze data on the Snowflake data cloud.

Snowpark Benefits:

With Snowpark, you can:

1. Write data processing code in your favorite programming language (other than SQL), such as Python, Java, or Scala.
2. Run your code on Snowflake’s servers, so you can take advantage of Snowflake’s scalability and performance.
3. Use Snowpark’s built-in libraries to perform common data processing tasks, such as data cleaning, transformation, and machine learning.

Use Cases for Snowpark

1. Data Integration: Snowpark can be used to build data pipelines that integrate data from multiple sources, such as databases, data lakes, and streaming sources.
2. Data Transformation: Snowpark can be used to transform data, such as cleaning and structuring data for analysis and reporting.
3. Data Processing: Snowpark can be used to process large volumes of data, such as aggregating and summarizing data for analysis.
4. Machine Learning: Snowpark can be used to build machine learning models that require large volumes of data and complex processing.

Connect Snowpark from your local machine

Prerequisites

1. You should have Python 3.8 installed on your machine.
2. You should have a Snowflake account (Trial accounts are fine too).

Step 1: Set up a virtual environment using Python 3.8.

Create a file requirements.txt with the following content.

snowflake-snowpark-python
snowflake-cli-labs
snowflake-snowpark-python[pandas]
config
black
isort

Create a PowerShell script file Set_Up_Python38_Virtual_Env.ps1 with the following content.

# Upgrade pip
python.exe -m pip install --upgrade pip
pip --version

# Install virtualenv
pip install virtualenv

# Set Python version to use
set-variable -name VERSION -value "38"

# Create a virtual environment
python -m virtualenv --python python$VERSION .venv
pause
# Press Shift + Control + P to open the Command Palette and click on the Python: Select Interpreter. Select the Python interpreter which came with your virtual environment.

# Activate environment and install dependency libraries
.venv\Scripts\activate.ps1
pip install -r requirements.txt

Execute the PowerShell file to set up a virtual environment with Python 3.8

Step 2: Execute a python script to connect to snowpark.

Establish a session with a Snowflake database using the same parameters (for example, the account name, user name, etc.) that you use in the connect function in the Snowflake Connector for Python.

Let’s save the following script as Snowpark.py and execute it —

from snowflake.snowpark.session import Session

def snowpark_session_create():
    connection_params = {
        "account": "xjb25065.us-east-1",
        "user": "<<YOUR SNOWFLAKE USERID>>",
        "password": "<<YOUR SNOWFLAKE PASSWORD>>",
        "role": "ACCOUNTADMIN",
        "warehouse": "COMPUTE_WH",
        "database": "MYTESTDB",
        "schema": "AWS_S3",
    }
    session = Session.builder.configs(connection_params).create()
    return session

test_session = snowpark_session_create()
print(test_session)

df = test_session.sql("SELECT * FROM SOURCE_DATA LIMIT 10")
df.show()

test_session.close()

Here, the DataFrame is utilized just to capture and accumulate the transformations. Once the accumulation is complete, it’s sent over to Snowflake (by calling either the show method or the collect method) for processing those transformations. Upon processing, Snowflake will send us back the results.

Step 3: The collect method.

The collect() method is used to process the rows one by one. Hence the following code will display the first names like below —

for row in df.collect():
    print(row.FIRST_NAME)

Step 4: Change the database and schema dynamically.

If you want to query a different database or schema, use the following options —

test_session.use_database("MYTESTDB")
test_session.use_schema("AWS_S3")

Execute various transformations on Snowpark dataframe

Apply filter conditions.

df = test_session.table("SOURCE_DATA")
df = df.select("FIRST_NAME", "LAST_NAME", "ADDRESS", "CITY", "STATE", "ZIP").filter(
    df.col("ZIP").between("8075", "9119")
)

df.show()

Calculate derived columns.

You can apply various functions on the existing columns to create new derived columns.

df = df.with_column("new_column", (df.col("A") - df.col("B") * df.col("C")))
df.show()

total_value = df.group_by("A").agg(df.sum("B").alias("C"))
total_value.show()

Convert snowpark dataframe to pandas dataframe and vice-versa.

# Converting snowpark dataframe to pandas dataframe
pandas_df = df.to_pandas()
pandas_df.head()

# Converting pandas dataframe to snowpark dataframe
snowpark_df = test_session.create_dataframe(pandas_df)
snowpark_df.show()

Rename a column.

df = df.with_column_renamed(df.col("old_column_name"),"new_column_name").show()

Join 2 dataframes.

df_A.join(df_B, df_A.key_col_name == df_B.key_col_name).select(
  df_A.col("X").alias("XX"),
  df_A.col("Y"),
  df_B.col("Z")
)

If you need to join a table with itself on different columns, you cannot perform the self-join with a single DataFrame. The join will fail because the column expressions for key column ("id") are present in the left and right sides of the join.

Instead, use Python’s built-in copy() method to create a clone of the DataFrame object, and use the two DataFrame objects to perform the join:

from copy import copy
from snowflake.snowpark.session import Session
from snowflake.snowpark.exceptions import SnowparkJoinException

def snowpark_session_create():
    connection_params = {
        "account": "xjb25065.us-east-1",
        "user": "dipan0saha",
        "password": "Dip@n0144",
        "role": "ACCOUNTADMIN",
        "warehouse": "COMPUTE_WH",
        "database": "MYTESTDB",
        "schema": "AWS_S3",
    }
    session = Session.builder.configs(connection_params).create()
    return session

test_session = snowpark_session_create()
print(test_session)

test_session.use_database("MYTESTDB")
test_session.use_schema("AWS_S3")

df_lhs = test_session.table("sample_product_data")
df_rhs = copy(df_lhs)

try:
    df_joined = df_lhs.join(df_rhs, df_lhs.col("id") == df_rhs.col("parent_id"))
except SnowparkJoinException as e:
    print(e.message)

print(df_joined.count())

test_session.close()

Drop a column.

df.drop(df.col("X")).show()

Snowflake conda channel

If you plan to use Anaconda (conda) instead, you can refer to this site for all the python packages which you can utilize.

Conclusion

Hope you found this blog useful! Let me know what you think about this, if you have any suggestion of a topic you would love to see here get in touch.

If you enjoyed the writings leave your claps 👏 to recommend this article so that others can see it.