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    doc_md=__doc__,
16    default_args=default_args,
17    tags=["example"],
18    schedule_interval="@daily",
19    catchup=False,
20) as dag:
21    # [START howto_operator_snowflake]
22    snowflake_op_sql_str = SnowflakeOperator(
23        task_id="snowflake_op_sql_str",
24        sql="CREATE TABLE IF NOT EXISTS test_table (id INT, name STRING);",
25        snowflake_conn_id="snowflake_default",
26    )
27
28    snowflake_op_with_params = SnowflakeOperator(
29        task_id="snowflake_op_with_params",
30        sql="INSERT INTO test_table (id, name) VALUES (%(id)s, %(name)s);",
31        parameters={"id": 1, "name": "test_name"},
32        snowflake_conn_id="snowflake_default",
33    )
34    # [END howto_operator_snowflake]
35
36    snowflake_op_sql_str >> snowflake_op_with_params