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",
8    start_date=datetime.datetime(2022, 1, 1),
9    schema="mysql_default",
10    catchup=False,
11) as dag:
12    # [START howto_operator_mysql]
13    drop_table_mysql_task = MySqlOperator(
14        task_id="drop_table_mysql",
15        sql=r"""DROP TABLE IF EXISTS table_name;""",
16    )
17
18    create_table_mysql_task = MySqlOperator(
19        task_id="create_table_mysql",
20        sql=r"""
21            CREATE TABLE table_name (
22                custom_id INTEGER NOT NULL,
23                user_name VARCHAR (50) NOT NULL,
24                timestamp TIMESTAMP NOT NULL
25            );
26        """,
27    )
28
29    insert_data_mysql_task = MySqlOperator(
30        task_id="insert_data_mysql",
31        sql=r"""
32            INSERT INTO table_name (custom_id, user_name, timestamp)
33            VALUES (1, 'user_1', '2022-01-01 00:00:00');
34        """,
35    )
36    # [END howto_operator_mysql]
37
38    drop_table_mysql_task >> create_table_mysql_task >> insert_data_mysql_task