airflow_mysql_operator_dag_create_table_insert_data.py

1import datetime
2
3from airflow import DAG
4from airflow.providers.mysql.operators.mysql import MySqlOperator
5
6with DAG(
7    dag_id="example_mysql_operator",
8    start_date=datetime.datetime(2022, 2, 2),
9    schedule="@daily",
10    catchup=False,
11) as dag:
12
13    # [START howto_operator_mysql]
14    create_table_mysql_task = MySqlOperator(
15        task_id="create_table_mysql",
16        mysql_conn_id="mysql_default",
17        sql=r"""
18            CREATE TABLE IF NOT EXISTS pet (
19                name VARCHAR(20) NOT NULL,
20                owner VARCHAR(20) NOT NULL,
21                species VARCHAR(20) NOT NULL,
22                sex CHAR(1),
23                birth DATE,
24                death DATE
25            );
26        """,
27    )
28    # [END howto_operator_mysql]
29
30    # [START howto_operator_mysql_external_file]
31    insert_data_mysql_task = MySqlOperator(
32        task_id="insert_data_mysql",
33        mysql_conn_id="mysql_default",
34        sql="INSERT INTO pet VALUES ('Puffball', 'Diane', 'hamster', 'f', '1999-03-30', NULL);",
35    )
36    # [END howto_operator_mysql_external_file]
37
38    create_table_mysql_task >> insert_data_mysql_task