Loading data into Snowflake is a common need. Using Python and pandas is a common go-to solution for data professionals. Whether you\u2019re pulling data from a relational database, wrangling a CSV file, or prototyping a new pipeline, this combination leverages pandas\u2019 intuitive data manipulation and Snowflake\u2019s cloud-native scalability. But let\u2019s be real\u2014data loading isn\u2019t always a simple task.<\/p>\n\n\n\n
Files go missing, connections drop, and type mismatches pop up when you least expect them. That\u2019s why robust error handling isn\u2019t just nice-to-have; it\u2019s essential for anything you\u2019d trust in production. In this guide, we\u2019ll walk through the fundamentals of getting data into Snowflake, explore practical examples with pandas and SQLAlchemy<\/a>, and equip you with the tools to build a dependable, real-world-ready pipeline. Let\u2019s dive in and make your data loading process as smooth as possible!<\/p>\n\n\n\n Before we dive into the code and examples, let’s chat about how Snowflake approaches data loading. Think of Snowflake like a really efficient postal service. Just as you might have different ways to send a package (regular mail, express delivery, or instant messenger), Snowflake offers various methods to load your data.<\/p>\n\n\n\n The “post office” in this case is what we call a staging area. You’ve got three types: Internal stages (like having a PO box at the post office) managed by Snowflake for simplified management; External stages that leverage cloud storage services like AWS S3, Azure Blob Storage, or GCP Cloud Storage; and User stages which provide personal file storage areas for individual users.<\/p>\n\n\n\n For File Processing:<\/strong> Snowflake automatically handles:<\/p>\n\n\n Snowflake’s automated type inference is powerful but needs careful attention. Initially, Snowflake analyzes a sample of rows to determine column data types. For example, for string columns, it considers the maximum length in the sample, not the entire dataset.<\/p>\n\n\n\n This is often good enough, but if the sample does not represent the longest values in a set, (e.g., one row with 200 characters among millions with 10), Snowflake might underestimate the required datatype length, leading to errors.<\/p>\n\n\n\n The best practice whenever possible, particularly when you process will be repeated or automated (like a production load), is to explicitly define column types to avoid type inference issues<\/p>\n\n\n\n Before we get into weeds of setting up python code, let’s explore different ways to load data into Snowflake based on common scenarios:<\/p>\n\n\n In this article, I will demonstrate something I do most every day- working with Python and pandas to load data into Snowflake using a few typical methods.<\/p>\n\n\n\n First, let’s look at a scenario where you’re loading customer transaction data from your e-commerce relational database. This example also covers how to write a pandas DataFrame to Snowflake using SQLAlchemy<\/a>, a Python SQL toolkit and Object Relational Mapper.<\/p>\n\n\n\n This is probably the most common scenario for data engineers, data scientists and analysts. This code snippet will feature:<\/p>\n\n\n In this example, we’ll extract yesterday’s transaction data from an e-commerce database (like MySQL or PostgreSQL) and load it into a Snowflake table called ‘daily_transactions’. The workflow includes handling potential errors, saving problematic records for further analysis, and implementing retry logic for temporary failures such as network timeouts.<\/p>\n\n\n\n The example assumes the source database has a ‘transactions’ table with a ‘date’ column (without a time of day aspect to the value) to filter by. The target Snowflake table structure will match the source data schema, as pandas will automatically create matching columns.<\/p>\n\n\n\n Let’s walk through this process step by step. First, we’ll connect to our source database and extract the previous day’s transaction data (this code can be downloaded from here<\/a>):<\/p>\n\n\n\n Next, we’ll set up our connection to Snowflake using SQLAlchemy. This provides a clean, standardized way to interact with Snowflake:<\/p>\n\n\n\n Finally, we’ll load the data into Snowflake with comprehensive error handling:<\/p>\n\n\n\n This will first attempt to execute the code in the try block to load the data into Snowflake and will be directed to the exception block in case of an error.<\/p>\n\n\n\n This snippet pulls yesterday’s transactions, connects to Snowflake, and loads the data in batches. If something goes wrong—like a timeout or a schema mismatch—it logs the error, saves the problematic records, and even tries again if the issue is transient. It’s a solid foundation for daily ETL jobs or ad-hoc analysis.<\/p>\n\n\n\n For optimal performance and security, consider these additional tips:<\/strong><\/p>\n\n\n The more practical application of this technique is for:<\/p>\n\n\n Now, let\u2019s shift gears to file-based data\u2014like a CSV with customer info. This scenario demands more: validating the data, ensuring consistency, and tracking what\u2019s loaded.<\/p>\n\n\n\n Let\u2019s see an example of how to load CSV Files with Custom Delimiter Detection into Snowflake and explore an approach to loading CSV files into Snowflake. The following class implements a comprehensive solution that:<\/p>\n\n\n This pattern is particularly useful for production ETL processes where reliability and traceability are critical. The code is structured as a reusable class that can be integrated into larger data pipelines (The code module described can be downloaded whole from here<\/a>):<\/p>\n\n\n\n Let\u2019s begin by importing python modules:<\/p>\n\n\n<\/a>Understanding How Snowflake Handles Data<\/h2>\n\n\n\n
\n
<\/a>Getting Your Hands Dirty: Different Ways to Load Data<\/h2>\n\n\n\n
\n
\n
\n
\n
<\/a>Using Python with pandas to load relational database data into Snowflake<\/h2>\n\n\n\n
\n
f_exists='append'<\/code> adds new data without overwriting existing records.<\/li>\n<\/ul>\n<\/div>\n\n\nimport pandas as pd\nfrom sqlalchemy import create_engine\nimport logging
from datetime import datetime\n\n# Connect to the source database and extract yesterday's transactions\n# In a production environment, you would configure this connection properly\n# and potentially include more sophisticated filtering logic\n\ntransactions_df = pd.read_sql(\n \"SELECT * FROM transactions WHERE date = CURRENT_DATE - 1\",\n source_database_connection\n # Note: 'date' here is a date field in the transactions table\n # This assumes a standard date column, but you might need to use\n # transaction_timestamp or another date\/time field in your database\n)<\/pre>\n\n\n\n# Create Snowflake connection using SQLAlchemy engine \n# For production environments, use environment variables or a secure credential store\nengine = create_engine(\n'snowflake:\/\/{user}:{password}@{account}\/{database}\/{schema}?warehouse={warehouse}&role={role}'.format(\n user='your_username',\n password='your_password',\n account='your_account',\n database='your_database',\n schema='your_schema',\n warehouse='your_warehouse',\n role='your_role'\n )\n)<\/pre>\n\n\n\n# Write to Snowflake (includes error handling for retries)\ntry:\n# The to_sql method will create or append to a table named 'daily_transactions' \n# The target table's schema will match our pandas DataFrame structure \n# Processing in chunks helps manage memory and enables better error recovery\n transactions_df.to_sql(\n 'daily_transactions',\n engine,\n if_exists='append', # Adds new records without overwriting existing data\n index=False, # Don't include DataFrame index as a column\n chunksize=10000 # Process in smaller batches for better memory management\n )\nexcept Exception as e:\n # Log the error with details\n error_message = f\"Failed to load transactions: {str(e)}\"\n logging.error(error_message)\n \n # Identify rows that aren't completely null (have at least one non-null value)\n # This helps isolate potentially problematic records\n failed_records = transactions_df[transactions_df.apply(lambda x: not x.isnull().all(), axis=1)]\n # Save failed records for review and later processing\nfailed_records.to_csv(f'failed_loads\/transactions_{datetime.now().strftime(\"%Y%m%d_%H%M%S\")}.csv')\n # Notify the team\n send_alert(error_message)\n # If the error is transient (like a network timeout or connection issue),\n # we can attempt to reload the data. The retry_load function would implement\n # exponential backoff to avoid overwhelming the server.\n if is_retryable_error(e):\n retry_load(transactions_df, max_retries=3)<\/pre>\n\n\n\n\n
\n
<\/a>Load File-Based Sources into Snowflake using Pandas<\/h2>\n\n\n\n
\n
\n
import pandas as pd\nfrom snowflake.connector.pandas_tools import write_pandas\nimport snowflake.connector\nimport logging\nfrom datetime import datetime\nimport os\nfrom typing import Optional, Tuple\nimport hashlib<\/pre>\n\n\n\n