airflow_dag_snowflake_operator_create_table_insert_data.py

1import datetime
2
3from airflow import DAG
4from airflow.providers.snowflake.operators.snowflake import SnowflakeOperator
5
6# These args will get passed on to each operator
7# You can override them on a per-task basis during operator initialization
8default_args = {
9    'owner': 'airflow',
10}
11
12with DAG(
13    dag_id='example_snowflake',
14    start_date=datetime.datetime(2021, 1, 1),
15    schema='public',
16    catchup=False,
17    default_args=default_args,
18) as dag:
19
20    snowflake_op_sql_str = SnowflakeOperator(
21        task_id='snowflake_op_sql_str',
22        snowflake_conn_id='snowflake_default',
23        sql="""
24            CREATE OR REPLACE TABLE fruit_table (
25                fruit_name VARCHAR(255),
26                quantity INTEGER
27            );
28            INSERT INTO fruit_table (fruit_name, quantity) VALUES ('apple', 10);
29            INSERT INTO fruit_table (fruit_name, quantity) VALUES ('banana', 20);
30        """,
31    )
32
33    snowflake_op_with_params = SnowflakeOperator(
34        task_id='snowflake_op_with_params',
35        snowflake_conn_id='snowflake_default',
36        sql="INSERT INTO fruit_table (fruit_name, quantity) VALUES (%(fruit_name)s, %(quantity)s);",
37        parameters={"fruit_name": "orange", "quantity": 15},
38    )
39
40    snowflake_op_sql_str >> snowflake_op_with_params