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 interactif Spark : vous devez cliquer 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. Recherchez 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 les pages apache-airflow-providers-apache-hive et apache-airflow-providers-common-sql.

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

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

   Paramètre	Description
   Connection Id	Le nom de la connexion. Exemple : adb_spark_cluster.
   Connection Type	Sélectionnez Hive Server 2 Thrift.
   Host	L'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	Le nom de la base de données. Exemple : adb_demo.
   Login	Le nom du groupe de ressources et le compte de base de données au format resource_group_name/database_account_name. Exemple : spark_interactive_prod/spark_user.
   Password	Le 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 la commande show databases selon une planification quotidienne :

   Paramètre	Obligatoire	Description
   task_id	Oui	L'ID de la tâche.
   conn_id	Oui	Le nom de la connexion créée à l'étape 4.
   sql	Oui	L'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 Airflow, cliquez sur à côté du DAG pour le déclencher.