How to use SnowflakeOperator with Airflow

artofdoe 0
Viewed 27

I have this super simple DAG with one task that's trying to execute a query on Snowflake

    default_args = {
    'owner': 'POC project',
    'depends_on_past': False,
    'start_date': datetime(2021, 6, 14),
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5)
}

 query_test = "select current_date() as date;"

 SNOWFLAKE_CONN_ID = """{
    "conn_type": "snowflake",
    "login": "zz__poc",
    "password": "123456",
    "schema": "users",
    "extra": {
        "account": "snf_account_poc",
        "database": "POC_DB",
        "region": "us-east",
        "warehouse": "POC_S_WH"
    }
}"""

dag = DAG(
    dag_id = 'SNOWFLAKE_QUERY',
    default_args=default_args,
    schedule_interval=timedelta(days=1),
    max_active_runs=1,
    catchup=False
)
#
# Connection Test
snowflake = SnowflakeOperator(
    task_id='test_snowflake_connection',
    sql=query_test,
    snowflake_conn_id=SNOWFLAKE_CONN_ID,
    dag=dag
)

I'm getting this following error:

    {standard_task_runner.py:107} ERROR - Failed to execute job 26 for task test_snowflake_connection (The conn_id `{
        "conn_type": "snowflake",
        "login": "zz__poc",
        "password": "123456",
        "schema": "users",
        "extra": {
            "account": "snf_account_poc",
            "database": "POC_DB",
            "region": "us-east",
            "warehouse": "POC_S_WH"
        }
}` isn't defined; 769

)

I'm pretty sure that all Snowflake credentials are correct. Can anyone give an example on how to properly use SnowflakeOperator or explain what the issue is with my code?

snowflake-cloud-data-platform airflow
1 Answers

snowflake_conn_id is expected to get the conn_id which will be used to retrieve the connection details from the backend table. It does not expect to get a full defined string of the connection and this is what the error tells you the id you passed is not defined.

To solve your issue you should define a snowflake connection. There are multiple ways you can do it see the docs for reference. For example you can do it from the UI via Admin -> Connections -> choose Snowflake, define the required parameters and assign conn_id to represent this connection.

Once the connection is defined you need simply to reference it from your DAG by using the conn_id you choose.

Elad Kalif0
Related