AnalyticDB:Use o Airflow para agendar jobs do Spark

Modo batch

Modo interativo

No modo interativo, o Airflow conecta-se ao cluster AnalyticDB for MySQL por meio de um endpoint JDBC usando HiveServer2 Thrift.

1. Obtenha o endpoint do grupo de recursos interativos do Spark. Clique em Apply for Endpoint ao lado de Public Endpoint para solicitar um endpoint público nos seguintes casos:

   • A ferramenta cliente usada para enviar jobs do Spark SQL está implantada em uma máquina local ou servidor externo.

   • A ferramenta cliente usada para enviar jobs do Spark SQL está implantada em uma instância ECS que não está na mesma VPC do cluster AnalyticDB for MySQL.

   1. Faça login no console do AnalyticDB for MySQL. No canto superior esquerdo, selecione uma região. No painel de navegação à esquerda, clique em Clusters e, em seguida, clique no ID do seu cluster.

   2. No painel de navegação, escolha Cluster Management > Resource Management e clique na aba Resource Groups.

   3. Localize o grupo de recursos desejado e clique em Details na coluna Actions. Copie o endpoint interno ou público. Você também pode copiar a string de conexão JDBC no campo Port.

2. Instale as dependências necessárias:

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

   Para detalhes sobre os pacotes, consulte apache-airflow-providers-apache-hive e apache-airflow-providers-common-sql.

3. Na interface web do Airflow, acesse Admin > Connections.

4. Clique em para adicionar uma conexão. Configure os seguintes parâmetros:

   Parâmetro	Descrição
   Connection Id	Nome da conexão. Exemplo: adb_spark_cluster.
   Connection Type	Selecione Hive Server 2 Thrift.
   Host	Endpoint obtido na etapa 1. Substitua default pelo nome real do banco de dados e remova o sufixo resource_group=<resource group name>. Exemplo: jdbc:hive2://amv-t4naxpqk****sparkwho.ads.aliyuncs.com:10000/adb_demo.
   Schema	Nome do banco de dados. Exemplo: adb_demo.
   Login	Nome do grupo de recursos e conta do banco de dados no formato resource_group_name/database_account_name. Exemplo: spark_interactive_prod/spark_user.
   Password	Senha da conta do banco de dados AnalyticDB for MySQL.
   Port	10000
   Extra	{"auth_mechanism": "CUSTOM"}

5. Crie um arquivo DAG. O exemplo abaixo executa show databases diariamente:

   Parâmetro	Obrigatório	Descrição
   task_id	Sim	ID do job.
   conn_id	Sim	Nome da conexão criada na etapa 4.
   sql	Sim	Instrução 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. Na interface web do Airflow, clique em ao lado do DAG para acioná-lo.