AnalyticDB:Orquestrar SQL do AnalyticDB for MySQL com Airflow

Pré-requisitos

• 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 de endereços IP do AnalyticDB for MySQL do cluster. Para mais detalhes, 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 esteja listado, execute o comando abaixo para instalar it:

      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 Admin > Connections.

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

      Parâmetro	Descrição
      Connection ID	ID exclusivo da conexão.
      Connection type	Selecione MySQL.
      Host	Endpoint do cluster AnalyticDB for MySQL. Consulte este endpoint na página Cluster Information no console.
      Login	Nome de usuário do AnalyticDB for MySQL.
      Password	Senha do AnalyticDB for MySQL.
      Port	Porta do cluster AnalyticDB for MySQL. 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 de 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 parâmetros.

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

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

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

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

   Após a execução da 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 da DAG.

   Na página Task Instances, visualize três registros de tarefas. A tarefa foi concluída com êxito se State for success, Dag ID for example_mysql, Task ID for mysql_test_task e Operator for MySqlOperator.

   Importante

   Por padrão, o Airflow usa o fuso horário UTC (Tempo Universal Coordenado). Portanto, o horário de execução exibido tem um atraso de 8 horas em relação ao Horário Padrão da China (UTC+8).