> ## Documentation Index
> Fetch the complete documentation index at: https://astronomer.io/docs/llms.txt
> Use this file to discover all available pages before exploring further.
# Orchestrate Snowflake Queries with Airflow
The key information from this and other Snowflake guides is available as an [Astronomer Cheat Sheet](https://www.astronomer.io/ebooks/airflow-snowflake-cheatsheet/?utm_source=website\&utm_medium=learn-guides\&utm_campaign=snowflake-tutorial).
[Snowflake](https://www.snowflake.com/) is one of the most commonly used data warehouses, and orchestrating Snowflake queries as part of a data pipeline is one of the most common Airflow use cases. Two Airflow provider packages, the [Snowflake Airflow provider](https://airflow.apache.org/docs/apache-airflow-providers-snowflake/stable/index.html) and the [Common SQL provider](https://airflow.apache.org/docs/apache-airflow-providers-common-sql/stable/index.html) contain hooks and operators that make it easy to interact with Snowflake from Airflow.
This tutorial covers an example of executing Snowflake operations with Airflow, including:
* Setting up a connection to Snowflake in Airflow.
* Executing individual SQL statements using the [`SQLExecuteQueryOperator`](https://airflow.apache.org/registry/providers/common-sql#common-sql-sql-SQLExecuteQueryOperator).
* Executing multiple SQL statements using the [`SnowflakeSqlApiOperator`](https://airflow.apache.org/registry/providers/snowflake#snowflake-snowflake-SnowflakeSqlApiOperator).
* Running data quality checks using the [`SQLColumnCheckOperator`](https://airflow.apache.org/registry/providers/common-sql#common-sql-sql-SQLColumnCheckOperator).
Additionally, [More on the Airflow Snowflake integration](#more-on-the-airflow-snowflake-integration) offers further information on general best practices and considerations when interacting with Snowflake from Airflow.
## Time to complete
This tutorial takes approximately 20 minutes to complete.
## Assumed knowledge
To get the most out of this tutorial, make sure you have an understanding of:
* Snowflake basics. See [Introduction to Snowflake](https://docs.snowflake.com/en/user-guide-intro.html).
* Airflow operators. See [Airflow operators](/docs/learn/what-is-an-operator).
* SQL basics. See the [W3 SQL tutorial](https://www.w3schools.com/sql/).
## Prerequisites
* The [Astro CLI](/docs/cli/v1.43/get-started-cli).
* A Snowflake account. A [30-day free trial](https://trial.snowflake.com/?owner=SPN-PID-365384) is available. You need to have at least one database, one schema and one warehouse set up in your Snowflake account as well as a user with the necessary permissions to create tables and run queries in the schema.
## Step 1: Configure your Astro project
Use the Astro CLI to create and run an Airflow project on your local machine.
1. Create a new Astro project:
```sh wrap theme={null}
$ mkdir astro-snowflake-tutorial && cd astro-snowflake-tutorial
$ astro dev init
```
2. In the `requirements.txt` file, add the [Snowflake Airflow provider](https://airflow.apache.org/docs/apache-airflow-providers-snowflake/stable/index.html) and the [Common SQL provider](https://airflow.apache.org/docs/apache-airflow-providers-common-sql/stable/index.html).
```txt wrap theme={null}
apache-airflow-providers-snowflake==6.4.0
apache-airflow-providers-common-sql==1.27.2
```
3. Run the following command to start your Airflow project:
```sh wrap theme={null}
astro dev start
```
## Step 2: Configure a Snowflake connection
There are different options to [authenticate to Snowflake](/docs/learn/connections/snowflake). The `SnowflakeAPIOperator` used in this tutorial requires you to use key-pair authentication, which is the preferred method. This method requires you to generate a public/private key pair, add the public key to your role in Snowflake, and use the private key in your [Airflow connection](/docs/learn/connections).
For Astro customers, Astronomer recommends taking advantage of the [Astro Environment Manager](/docs/astro/manage-connections-variables#astro-environment-manager) to store connections in an Astro-managed secrets backend. These connections can be shared across multiple deployed and local Airflow environments. See [Manage Astro connections in branch-based deploy workflows](/docs/astro/best-practices/connections-branch-deploys).
1. In your terminal, run the following command to [generate a private RSA key using OpenSSL](https://docs.openssl.org/master/man1/openssl-genrsa/). Note that while there are other options to generate a key pair, Snowflake has [specific requirements for the key format](https://docs.snowflake.com/en/user-guide/key-pair-auth) and may not accept keys generated with other tools. Make sure to write down the key passphrase as you will need it later.
```bash wrap theme={null}
openssl genrsa 2048 | openssl pkcs8 -topk8 -v2 des3 -inform PEM -out rsa_key.p8
```
2. Generate the associated public key using the following command:
```bash wrap theme={null}
openssl rsa -in rsa_key.p8 -pubout -out rsa_key.pub
```
3. Format the private key. Version 6.3.0+ of the Airflow Snowflake provider requires the private key to be base64 encoded. You can create a base64 encoded key with the following script:
```python wrap theme={null}
import base64
with open("path/to/rsa_key.p8", "rb") as key_file:
private_key_content = base64.b64encode(key_file.read()).decode("utf-8")
print(private_key_content)
```
If you're on version 6.2.2 or older of the Airflow Snowflake provider, you need to provide the private key without any coding conversions but with newlines encoded as `\n`. You can use the script below to format the key correctly:
```python wrap theme={null}
def format_private_key(private_key_path):
with open(private_key_path, 'r') as key_file:
private_key = key_file.read()
return private_key.replace('\n', '\\n')
formatted_key = format_private_key('rsa_key.pem')
print(formatted_key)
```
4. In the Snowflake UI, run the following SQL command to add the **public** key to your [user](https://docs.snowflake.com/en/user-guide/admin-user-management). You can paste the **public** key directly from the `rsa_key.pub` file without needing to modify it.
```sql wrap theme={null}
ALTER USER SET RSA_PUBLIC_KEY='';
```
5. In the Airflow UI, go to **Admin** -> **Connections** and click **+** to create a new connection. Choose the `Snowflake` connection type and enter the following information:
* **Connection ID**: `snowflake_conn`
* **[Schema](https://docs.snowflake.com/en/sql-reference/sql/create-schema.html)**: Your Snowflake schema. The example DAG uses `DEMO_SCHEMA`.
* **Login**: Your Snowflake [user](https://docs.snowflake.com/en/sql-reference/sql/create-user) name. Make sure to capitalize the user name as the `SnowflakeAPIOperator` requires it.
* **Password**: Your private key passphrase.
* **Extra**: Enter the following JSON object with your own Snowflake [account identifier](https://docs.snowflake.com/en/user-guide/admin-account-identifier), [database](https://docs.snowflake.com/en/sql-reference/sql/create-database), your [role](https://docs.snowflake.com/en/sql-reference/sql/create-role) in properly capitalized format, and your [warehouse](https://docs.snowflake.com/en/sql-reference/sql/create-warehouse).
```json wrap theme={null}
{
"account": "",
"warehouse": "",
"database": "DEMO_DB",
"region": "",
"role": "",
"private_key_content": "LS0....C0="
}
```
When using JSON format to set your connection, use the following parameters:
```json wrap theme={null}
AIRFLOW_CONN_SNOWFLAKE_DEFAULT='{
"conn_type":"snowflake",
"login":"",
"password":"",
"schema":"DEMO_SCHEMA",
"extra":{
"account":"",
"database":"DEMO_DB",
"region":"",
"role":"",
"private_key_content":"LS0....C0="
}
}'
```
## Step 3: Add your SQL statements
The DAG you will create in Step 4 runs multiple SQL statements against your Snowflake data warehouse. While it is possible to add SQL statements directly in your DAG file it is common practice to store them in separate files. When initializing your Astro project with the Astro CLI, an `include` folder was created. The contents of this folder will automatically be mounted into the Dockerfile, which makes it the standard location in which supporting files are stored.
1. Create a folder called `sql` in your `include` folder.
2. Create a new file in `include/sql` called `insert_data.sql` and copy the following code:
```sql wrap theme={null}
INSERT INTO {{ params.db_name }}.{{ params.schema_name }}.{{ params.table_name }} (ID, NAME)
VALUES
(1, 'Avery');
```
This file contains one SQL statement that inserts a row into a table. The database, schema, and table names are parameterized so that you can pass them to the operator at runtime.
3. The `SnowflakeSqlApiOperator` can run multiple SQL statements in a single task. Create a new file in `include/sql` called `multiple_statements_query.sql` and copy the following code:
```sql wrap theme={null}
INSERT INTO {{ params.db_name }}.{{ params.schema_name }}.{{ params.table_name }} (ID, NAME)
VALUES
(2, 'Peanut'),
(3, 'Butter');
INSERT INTO {{ params.db_name }}.{{ params.schema_name }}.{{ params.table_name }} (ID, NAME)
VALUES
(4, 'Vega'),
(5, 'Harper');
```
This file contains two SQL statements that insert multiple rows into a table.
When running SQL statements from Airflow operators, you can store the SQL code in individual SQL files, in a combined SQL file, or as strings in a Python module. Astronomer recommends storing lengthy SQL statements in a dedicated file to keep your DAG files clean and readable.
## Step 4: Write a Snowflake DAG
1. Create a new file in your `dags` directory called `my_snowflake_dag.py`.
2. Copy and paste the code below into the file:
```python expandable wrap theme={null}
"""
### Snowflake Tutorial DAG
This DAG demonstrates how to use the SQLExecuteQueryOperator,
SnowflakeSqlApiOperator and SQLColumnCheckOperator to interact with Snowflake.
"""
from airflow.decorators import dag
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator
from airflow.providers.common.sql.operators.sql import SQLColumnCheckOperator
from airflow.providers.snowflake.operators.snowflake import SnowflakeSqlApiOperator
from airflow.models.baseoperator import chain
from pendulum import datetime, duration
import os
_SNOWFLAKE_CONN_ID = "snowflake_conn"
_SNOWFLAKE_DB = "DEMO_DB"
_SNOWFLAKE_SCHEMA = "DEMO_SCHEMA"
_SNOWFLAKE_TABLE = "DEMO_TABLE"
@dag(
dag_display_name="Snowflake Tutorial DAG ❄️",
start_date=datetime(2024, 9, 1),
schedule=None,
catchup=False,
default_args={"owner": "airflow", "retries": 1, "retry_delay": duration(seconds=5)},
doc_md=__doc__,
tags=["tutorial"],
template_searchpath=[
os.path.join(os.path.dirname(os.path.abspath(__file__)), "../include/sql")
], # path to the SQL templates
)
def my_snowflake_dag():
# you can execute SQL queries directly using the SQLExecuteQueryOperator
create_or_replace_table = SQLExecuteQueryOperator(
task_id="create_or_replace_table",
conn_id=_SNOWFLAKE_CONN_ID,
database="DEMO_DB",
sql=f"""
CREATE OR REPLACE TABLE {_SNOWFLAKE_SCHEMA}.{_SNOWFLAKE_TABLE} (
ID INT,
NAME VARCHAR
)
""",
)
# you can also execute SQL queries from a file, make sure to add the path to the template_searchpath
insert_data = SQLExecuteQueryOperator(
task_id="insert_data",
conn_id=_SNOWFLAKE_CONN_ID,
database="DEMO_DB",
sql="insert_data.sql",
params={
"db_name": _SNOWFLAKE_DB,
"schema_name": _SNOWFLAKE_SCHEMA,
"table_name": _SNOWFLAKE_TABLE,
},
)
# you can also execute multiple SQL statements using the SnowflakeSqlApiOperator
# make sure to set the statement_count parameter to the number of statements in the SQL file
# and that your connection details are in their proper capitalized form!
insert_data_multiple_statements = SnowflakeSqlApiOperator(
task_id="insert_data_multiple_statements",
snowflake_conn_id=_SNOWFLAKE_CONN_ID,
sql="multiple_statements_query.sql",
database=_SNOWFLAKE_DB,
schema=_SNOWFLAKE_SCHEMA,
params={
"db_name": _SNOWFLAKE_DB,
"schema_name": _SNOWFLAKE_SCHEMA,
"table_name": _SNOWFLAKE_TABLE,
},
statement_count=2, # needs to match the number of statements in the SQL file
autocommit=True,
)
# use SQLCheck operators to check the quality of your data
data_quality_check = SQLColumnCheckOperator(
task_id="data_quality_check",
conn_id=_SNOWFLAKE_CONN_ID,
database=_SNOWFLAKE_DB,
table=f"{_SNOWFLAKE_SCHEMA}.{_SNOWFLAKE_TABLE}",
column_mapping={
"ID": {"null_check": {"equal_to": 0}, "distinct_check": {"geq_to": 3}}
},
)
chain(
create_or_replace_table,
insert_data,
insert_data_multiple_statements,
data_quality_check,
)
my_snowflake_dag()
```
The DAG completes the following steps:
* Uses the [`SQLExecuteQueryOperator`](https://airflow.apache.org/registry/providers/common-sql#common-sql-sql-SQLExecuteQueryOperator) to run an in-line SQL statement that creates a table in Snowflake.
* Uses the `SQLExecuteQueryOperator` to run an SQL file containing a singular SQL statement that inserts data into the table.
* Uses the [`SnowflakeSqlApiOperator`](https://airflow.apache.org/registry/providers/snowflake#snowflake-snowflake-SnowflakeSqlApiOperator) to run an SQL file containing multiple SQL statements that insert data into the table. The operator is set to run in [deferrable](/docs/learn/deferrable-operators) mode.
* Uses the [`SQLColumnCheckOperator`](https://airflow.apache.org/registry/providers/common-sql#common-sql-sql-SQLColumnCheckOperator) to run a data quality check on the table checking that there are no NULL values in the `ID` column and that it contains at least 3 distinct values. To learn more about SQL check operators, see [Run data quality checks using SQL check operators](/docs/learn/airflow-sql-data-quality).
The `chain()` method at the end of the DAG sets the [dependencies](/docs/learn/managing-dependencies). This method is commonly used over bitshift operators (`>>`) to make it easier to read dependencies between many tasks.
## Step 5: Run the DAG
1. In the Airflow UI, click the play button to manually run your DAG.
2. Open the [logs](/docs/learn/logging) for the `data_quality_check` task to see the results of the data quality check, confirming that the table was created and populated correctly.
```text wrap theme={null}
[2025-07-03, 16:03:09 UTC] {sql.py:469} INFO - Record: [('ID', 'null_check', 0), ('ID', 'distinct_check', 5)]
[2025-07-03, 16:03:09 UTC] {sql.py:492} INFO - All tests have passed
```
## More on the Airflow Snowflake integration
This section provides additional information on orchestrating actions in Snowflake with Airflow.
### Snowflake operators and hooks
Several open source packages contain operators used to orchestrate Snowflake in Airflow.
The [Common SQL provider package](https://airflow.apache.org/registry/providers/common-sql) contains operators that you can use with several SQL databases, including Snowflake:
* [`SQLExecuteQueryOperator`](https://airflow.apache.org/registry/providers/common-sql#common-sql-sql-SQLExecuteQueryOperator): Executes a single SQL statement. This operator replaces the deprecated `SnowflakeOperator`.
* [`SQLColumnCheckOperator`](https://airflow.apache.org/registry/providers/common-sql#common-sql-sql-SQLColumnCheckOperator): Performs a data quality check against columns of a given table. See [Run data quality checks using SQL check operators](/docs/learn/airflow-sql-data-quality).
* [`SQLTableCheckOperator`](https://airflow.apache.org/registry/providers/common-sql#common-sql-sql-SQLTableCheckOperator): Performs a data quality check against a given table.
The [Snowflake provider package](https://airflow.apache.org/registry/providers/snowflake/) contains:
* [`SnowflakeSqlApiOperator`](https://airflow.apache.org/registry/providers/snowflake#snowflake-snowflake-SnowflakeSqlApiOperator): Executes multiple SQL statements in a single task. Note that this operator uses the Snowflake SQL API, which requires connection parameters such as the role and user to be properly capitalized. The operator can be set to be [deferrable](/docs/learn/deferrable-operators) using `deferrable=True`.
* [`CopyFromExternalStageToSnowflakeOperator`](https://airflow.apache.org/registry/providers/snowflake#snowflake-copy_into_snowflake-CopyFromExternalStageToSnowflakeOperator): Copies data from an external stage to a Snowflake table. Note that the `prefix` parameter will be added to the full stage path defined in Snowflake.
* [`SnowflakeHook`](https://airflow.apache.org/registry/providers/snowflake#snowflake-snowflake-SnowflakeHook): A client to interact with Snowflake which is commonly used when building custom operators interacting with Snowflake.
### Best practices and considerations
The following are some best practices and considerations to keep in mind when orchestrating Snowflake queries from Airflow:
* To reduce costs and improve the scalability of your Airflow environment, consider using the `SnowflakeSqlApiOperator` in [deferrable](/docs/learn/deferrable-operators) mode for long running queries.
* Set your default Snowflake query specifications such as Warehouse, Role, Schema, and so on in the Airflow connection. Then overwrite those parameters for specific tasks as necessary in your operator definitions. This is cleaner and easier to read than adding `USE Warehouse XYZ;` statements within your queries. If you are an Astro customer, use the [Astro Environment Manager](/docs/astro/create-and-link-connections) to define your base connection and add overrides for specific deployments and tasks.
* Pay attention to which Snowflake compute resources your tasks are using, as overtaxing your assigned resources can cause slowdowns in your Airflow tasks. It is generally recommended to have different warehouses devoted to your different Airflow environments to ensure DAG development and testing doesn't interfere with DAGs running in production. If you want to optimize your Snowflake usage, consider using [SnowPatrol](/docs/learn/reference-architecture-snowpatrol) to detect anomalies in your Snowflake spend.
* Make use of [Snowflake stages](https://docs.snowflake.com/en/sql-reference/sql/create-stage.html) together with the [`CopyFromExternalStageToSnowflakeOperator`](https://airflow.apache.org/registry/providers/snowflake#snowflake-copy_into_snowflake-CopyFromExternalStageToSnowflakeOperator) when loading large amounts data from an external system using Airflow.
## Conclusion
Congratulations! You've connected Airflow to Snowflake and executed Snowflake queries from your Airflow DAGs. You've also learned about best practices and considerations when orchestrating Snowflake queries from Airflow.