AnalyticDB:Utiliser Airflow pour planifier des tâches Spark

Mode batch

Mode interactif

Le mode interactif connecte Airflow au cluster AnalyticDB for MySQL via un endpoint JDBC utilisant HiveServer2 Thrift.

1. Obtenez l'endpoint du groupe de ressources interactives Spark : cliquez sur Apply for Endpoint à côté de Public Endpoint pour demander un endpoint public dans les cas suivants :

   • L'outil client utilisé pour soumettre des tâches Spark SQL est déployé sur une machine locale ou un serveur externe.

   • L'outil client utilisé pour soumettre des tâches Spark SQL est déployé sur une instance ECS, et l'instance ECS et le cluster AnalyticDB for MySQL ne se trouvent pas dans le même VPC.

   1. Connectez-vous à la console AnalyticDB for MySQL. Dans le coin supérieur gauche, sélectionnez une région. Dans le volet de navigation de gauche, cliquez sur Clusters, puis cliquez sur l'ID de votre cluster.

   2. Dans le volet de navigation, choisissez Cluster Management > Resource Management, puis cliquez sur l'onglet Resource Groups.

   3. Localisez le groupe de ressources cible et cliquez sur Details dans la colonne Actions. Copiez l'endpoint interne ou public. Vous pouvez également copier la chaîne de connexion JDBC depuis le champ Port.

2. Installez les dépendances requises :

   pip install apache-airflow-providers-apache-hive "apache-airflow-providers-common-sql==1.21.0"

   Pour plus de détails sur les packages, consultez apache-airflow-providers-apache-hive et apache-airflow-providers-common-sql.

3. Dans l'interface web d'Airflow, accédez à Admin > Connections.

4. Cliquez sur pour ajouter une connexion. Configurez les paramètres suivants :

   Paramètre	Description
   Connection Id	Nom de la connexion. Exemple : adb_spark_cluster.
   Connection Type	Sélectionnez Hive Server 2 Thrift.
   Host	Endpoint obtenu à l'étape 1. Remplacez default par le nom réel de la base de données et supprimez le suffixe resource_group=<resource group name>. Exemple : jdbc:hive2://amv-t4naxpqk****sparkwho.ads.aliyuncs.com:10000/adb_demo.
   Schema	Nom de la base de données. Exemple : adb_demo.
   Login	Nom du groupe de ressources et compte de base de données au format resource_group_name/database_account_name. Exemple : spark_interactive_prod/spark_user.
   Password	Mot de passe du compte de base de données AnalyticDB for MySQL.
   Port	10000
   Extra	{"auth_mechanism": "CUSTOM"}

5. Créez un fichier DAG. L'exemple suivant exécute show databases selon une planification quotidienne :

   Paramètre	Obligatoire	Description
   task_id	Oui	ID de la tâche.
   conn_id	Oui	Nom de la connexion défini à l'étape 4.
   sql	Oui	Instruction Spark SQL.

   from airflow import DAG
   from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator
   from datetime import datetime
   
   default_args = {
       'owner': 'airflow',
       'start_date': datetime(2025, 2, 10),
       'retries': 1,
   }
   
   dag = DAG(
       'adb_spark_sql_test',
       default_args=default_args,
       schedule_interval='@daily',
   )
   
   jdbc_query = SQLExecuteQueryOperator(
       task_id='execute_spark_sql_query',
       conn_id='adb_spark_cluster',  # Connection created in step 4
       sql='show databases',
       dag=dag
   )
   
   jdbc_query

6. Dans l'interface web d'Airflow, cliquez sur à côté du DAG pour le déclencher.