AnalyticDB:Orchestrate AnalyticDB for MySQL SQL with Airflow

Prerequisites

• An AnalyticDB for MySQL Enterprise Edition, Basic Edition, or Data Lakehouse Edition cluster is created.

• Airflow is installed. For more information, see the Airflow documentation.

• Ensure the Airflow server's IP address is on the AnalyticDB for MySQL of the AnalyticDB for MySQL cluster. For more information, see Configure an IP address whitelist.

Procedure

1. Check whether the apache-airflow-providers-mysql provider is installed.

   1. In the Airflow UI, click Admin > Providers.

   2. On the Providers page, check whether apache-airflow-providers-mysql is in the list.

   3. (Optional) If the apache-airflow-providers-mysql provider is not in the list, run the following command to install it:

      pip install apache-airflow-providers-mysql

      Important

      If the OSError: mysql_config not found error occurs, run the yum install mysql-devel command to install MySQL development files. Then, run the installation command for apache-airflow-providers-mysql again.

2. Create a connection.

   1. In the Airflow UI, click Admin > Connections.

   2. Click the icon. On the Add Connection page, configure the following parameters.

      Parameter	Description
      Connection ID	A unique ID for the connection.
      Connection type	Select MySQL.
      Host	The endpoint of the AnalyticDB for MySQL cluster. Find this endpoint on the Cluster Information page in the console.
      Login	The username of the AnalyticDB for MySQL.
      Password	The password of the AnalyticDB for MySQL.
      Port	The AnalyticDB for MySQL of the AnalyticDB for MySQL cluster. The value is fixed at 3306.

      Note

      Other parameters are optional. Configure them as needed.

3. Go to the Airflow installation directory and check the dags_folder parameter in the airflow.cfg file.

   1. Go to the Airflow installation directory.

      cd /root/airflow

   2. Check the dags_folder parameter in the airflow.cfg file.

      cat airflow.cfg

   3. (Optional) If no folder exists in the path specified by the dags_folder parameter, run the mkdir command to create the folder.

      Note

      For example, if the path for dags_folder is /root/airflow/dags but the dags folder does not exist in the /root/airflow directory, create it.

4. Create a DAG file, such as mysql_dags.py:

   from airflow import DAG
   from airflow.providers.mysql.operators.mysql import MySqlOperator
   from airflow.utils.dates import days_ago
   default_args = {
       'owner': 'airflow',
   }
   dag = DAG(
       'example_mysql',
       default_args=default_args,
       start_date=days_ago(2),
       tags=['example'],
   )
   mysql_test = MySqlOperator(
       task_id='mysql_test',
       mysql_conn_id='test',
       sql='SHOW DATABASES;',
       dag=dag,
   )
   mysql_test_task = MySqlOperator(
       task_id='mysql_test_task',
       mysql_conn_id='test',
       sql='SELECT * FROM test;',
       dag=dag,
   )
   mysql_test >> mysql_test_task
   if __name__ == "__main__":
       dag.cli()
   

   The following table describes the parameters.

   • mysql_conn_id: The ID of the connection that you created in Step 2.

   • sql: the SQL statement that you want to execute.

   For more information about the parameters, see the Airflow documentation.

5. In the Airflow UI, find your DAG and click the icon in the Actions column to run it.

   After the DAG runs, click the green circle in the Runs column to view the run details.

   A "1" in the green circle next to example_mysql indicates one successful DAG run.

   On the Task Instances page, you can view three task records. If State is success, Dag ID is example_mysql, Task ID is mysql_test_task, and Operator is MySqlOperator, the task ran successfully.

   Important

   By default, Airflow uses the Coordinated Universal Time (UTC) time zone. This means the displayed execution time is 8 hours behind China Standard Time (UTC+8).