airflow_snowflake_operator_dag_create_table_insert_data.py

1import datetime
2
3from airflow import DAG
4from airflow.providers.snowflake.operators.snowflake import SnowflakeOperator
5
6# Define the SQL commands to be executed
7CREATE_TABLE_SQL_STRING = (
8    "CREATE OR REPLACE TABLE public.test_table (id int, name string);"
9)
10INSERT_DATA_SQL_STRING = (
11    "INSERT INTO public.test_table VALUES (1, 'airflow'), (2, 'snowflake');"
12)
13
14with DAG(
15    dag_id='example_snowflake',
16    start_date=datetime.datetime(2023, 1, 1),
17    schedule_interval='@daily',
18    catchup=False,
19) as dag:
20
21    # Task to create a table in Snowflake
22    create_table = SnowflakeOperator(
23        task_id='create_table',
24        sql=CREATE_TABLE_SQL_STRING,
25        snowflake_conn_id='snowflake_default',
26    )
27
28    # Task to insert data into the created table
29    insert_data = SnowflakeOperator(
30        task_id='insert_data',
31        sql=INSERT_DATA_SQL_STRING,
32        snowflake_conn_id='snowflake_default',
33    )
34
35    create_table >> insert_data