AnalyticDB:Orchestrate AnalyticDB for MySQL SQL with Airflow

Prérequis

• Un cluster Enterprise Edition, Basic Edition ou Data Lakehouse Edition AnalyticDB for MySQL est créé.

• Airflow est installé. Pour plus d'informations, consultez la documentation Airflow.

• L'adresse IP du serveur Airflow est ajoutée à la liste d'autorisation du cluster AnalyticDB for MySQL. Pour plus d'informations, consultez la rubrique Configurer une liste d'autorisation d'adresses IP.

Procédure

1. Vérifiez si le fournisseur apache-airflow-providers-mysql est installé.

   1. Dans l'interface utilisateur Airflow, cliquez sur Admin > Providers.

   2. Sur la page Providers, vérifiez la présence de apache-airflow-providers-mysql dans la liste.

   3. (Facultatif) Si le fournisseur apache-airflow-providers-mysql ne figure pas dans la liste, exécutez la commande suivante pour l'installer :

      pip install apache-airflow-providers-mysql

      Important

      Si l'erreur OSError: mysql_config not found se produit, exécutez la commande yum install mysql-devel pour installer les fichiers de développement MySQL. Ensuite, relancez la commande d'installation de apache-airflow-providers-mysql.

2. Créez une connexion.

   1. Dans l'interface utilisateur Airflow, cliquez sur Admin > Connections.

   2. Cliquez sur l'icône . Sur la page Add Connection, configurez les paramètres suivants.

      Parameter	Description
      Connection ID	ID unique de la connexion.
      Connection type	Sélectionnez MySQL.
      Host	Endpoint du cluster AnalyticDB for MySQL. Vous trouverez cet endpoint sur la page Cluster Information de la console.
      Login	Nom d'utilisateur du compte AnalyticDB for MySQL.
      Password	Mot de passe du compte AnalyticDB for MySQL.
      Port	Port du cluster AnalyticDB for MySQL. La valeur est fixée à 3306.

      Remarque

      Les autres paramètres sont facultatifs. Configurez-les selon vos besoins.

3. Accédez au répertoire d'installation d'Airflow et vérifiez le paramètre dags_folder dans le fichier airflow.cfg.

   1. Accédez au répertoire d'installation d'Airflow.

      cd /root/airflow

   2. Vérifiez le paramètre dags_folder dans le fichier airflow.cfg.

      cat airflow.cfg

   3. (Facultatif) Si le dossier spécifié par le paramètre dags_folder n'existe pas, exécutez la commande mkdir pour le créer.

      Remarque

      Par exemple, si le chemin dags_folder est /root/airflow/dags mais que le dossier dags n'existe pas dans le répertoire /root/airflow, créez-le.

4. Créez un fichier DAG, tel que 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()
   

   Le tableau suivant décrit les paramètres clés.

   • mysql_conn_id : ID de connexion créé à l'étape 2.

   • sql : instruction SQL à exécuter.

   Pour plus d'informations sur ces paramètres, consultez la documentation Airflow.

5. Dans l'interface utilisateur Airflow, localisez votre DAG et cliquez sur l'icône dans la colonne Actions pour l'exécuter.

   Une fois l'exécution du DAG terminée, cliquez sur le cercle vert dans la colonne Runs pour afficher les détails de l'exécution.

   Un « 1 » dans le cercle vert à côté de example_mysql indique une exécution réussie du DAG.

   Sur la page Task Instances, vous pouvez consulter trois enregistrements de tâches. Si le champ State affiche success, que le Dag ID est example_mysql, que le Task ID est mysql_test_task et que l'Operator est MySqlOperator, la tâche a été exécutée avec succès.

   Important

   Par défaut, Airflow utilise le fuseau horaire UTC (Coordinated Universal Time). Cela signifie que l'heure d'exécution affichée accuse un retard de 8 heures par rapport à l'heure normale de Chine (UTC+8).