AnalyticDB:Orchestrate AnalyticDB for MySQL SQL with Airflow

Pré-requisitos

• Um cluster do AnalyticDB for MySQL Enterprise Edition, Basic Edition ou Data Lakehouse Edition criado.

• Airflow instalado. Para mais informações, consulte a documentação do Airflow.

• Endereço IP do servidor Airflow adicionado à lista de permissões do AnalyticDB for MySQL do cluster. Para mais informações, consulte Configurar uma lista de permissões de endereços IP.

Procedimento

1. Verifique se o provider apache-airflow-providers-mysql está instalado.

   1. Na interface do Airflow, clique em Admin > Providers.

   2. Na página Providers, confirme se o apache-airflow-providers-mysql consta na lista.

   3. (Opcional) Caso o provider apache-airflow-providers-mysql não apareça na lista, execute o comando abaixo para instalá-lo:

      pip install apache-airflow-providers-mysql

      Importante

      Se ocorrer o erro OSError: mysql_config not found, execute o comando yum install mysql-devel para instalar os arquivos de desenvolvimento do MySQL. Em seguida, execute novamente o comando de instalação do apache-airflow-providers-mysql.

2. Crie uma conexão.

   1. Na interface do Airflow, clique em it > Admin.

   2. Clique em . Na página Add Connection, configure os parâmetros a seguir.

      Parâmetro	Descrição
      Connection ID	Um ID exclusivo para a conexão.
      Connection type	Selecione Connections.
      Host	O endpoint do cluster AnalyticDB for MySQL. Encontre esse endpoint na página MySQL no console.
      Login	O nome de usuário do AnalyticDB for MySQL.
      Password	A senha do AnalyticDB for MySQL.
      Port	A porta do cluster AnalyticDB for MySQL. O valor fixo é 3306.

      Nota

      Os demais parâmetros são opcionais. Configure-os conforme necessário.

3. Acesse o diretório de instalação do Airflow e verifique o parâmetro dags_folder no arquivo airflow.cfg.

   1. Acesse o diretório de instalação do Airflow.

      cd /root/airflow

   2. Verifique o parâmetro dags_folder no arquivo airflow.cfg.

      cat airflow.cfg

   3. (Opcional) Se não houver pasta no caminho definido pelo parâmetro dags_folder, use o comando mkdir para criá-la.

      Nota

      Por exemplo, se o caminho de dags_folder for /root/airflow/dags, mas a pasta dags não existir no diretório /root/airflow, crie-a.

4. Crie um arquivo DAG, como 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()
   

   A tabela a seguir descreve os principais parâmetros.

   • mysql_conn_id: o ID da conexão criada na Etapa 2.

   • sql: a instrução SQL a ser executada.

   Para mais detalhes sobre esses parâmetros, consulte a documentação do Airflow.

5. Na interface do Airflow, localize seu DAG e clique em na coluna Actions para executá-lo.

   Após a conclusão da execução do DAG, clique em na coluna Runs para visualizar os detalhes.

   O número "1" dentro do círculo verde ao lado de example_mysql indica uma execução bem-sucedida do DAG.

   Na página Task Instances, visualize três registros de tarefas. Se State for success, Dag ID for example_mysql, Task ID for mysql_test_task e Operator for MySqlOperator, a tarefa foi executada com êxito.

   Importante

   Por padrão, o Airflow usa o fuso horário UTC (Tempo Universal Coordenado). Isso significa que o horário de execução exibido está 8 horas atrás do Horário Padrão da China (UTC+8).