How to pass snowflake connection parameters in snowflake operator in airflow

In order to pass connection parameters to the SnowflakeOperator in Airflow, you need to first create a connection object in the Airflow UI. This connection object should contain all the necessary information to connect to your Snowflake instance, such as the user, password, account, and warehouse.

Here is an example of how you would create a connection object in Airflow:

1. Go to the Airflow UI, and navigate to the “Admin” menu.
2. Select “Connections” from the submenu.
3. In the “Connections” page, select “Create” to create a new connection object.
4. Fill in the form with the necessary connection information. For example, for a Snowflake connection, you would need to provide the following information:

• Conn Id: A unique identifier for the connection.
• Conn Type: “Snowflake”.
• Host: The hostname or IP address of your Snowflake instance.
• Login: The username to use when connecting to Snowflake.
• Password: The password for the user.
• Schema: The default schema to use when querying Snowflake.
• Warehouse: The warehouse to use when querying Snowflake.

5. Once you have filled in the form, click “Save” to create the connection object.

Once the connection object is created, you can use the “conn_id” parameter when instantiating the SnowflakeOperator to specify which connection to use. Here is an example of how you would use the SnowflakeOperator to run a query:

from airflow.operators.snowflake_operator import SnowflakeOperator

query = "SELECT viewer_id,viewer_cnt, as_on_date FROM freshers_in"

snowflake_task = SnowflakeOperator(
    task_id='snowflake_task',
    sql=query,
    snowflake_conn_id='my_conn_id',
    autocommit=True,
    warehouse='my_warehouse'
)

freshers_in_snowflake_op_sql_str = SnowflakeOperator(task_id="snowflake_op_sql_str", sql=CREATE_TABLE_SQL_STRING)

freshers_in_snowflake_op_with_params = SnowflakeOperator(
    task_id="freshers_in_snowflake_op_with_params",
    sql=SQL_INSERT_STATEMENT,
    parameters={"id": 56},
)

freshers_in_snowflake_op_sql_list = SnowflakeOperator(task_id="freshers_in_snowflake_op_sql_list", sql=SQL_LIST)

snowflake_op_sql_multiple_stmts = SnowflakeOperator(
    task_id="snowflake_op_sql_multiple_stmts",
    sql=SQL_MULTIPLE_STMTS,
    split_statements=True,
)

freshers_in_snowflake_op_template_file = SnowflakeOperator(
    task_id="freshers_in_snowflake_op_template_file",
    sql="example_snowflake_freshers_in_snowflake_op_template_file.sql",
)