¶ Configurar conexão MySQL no Apache Airflow
Guia rápido e prático para deixar seu Airflow falando com um banco MySQL. Inclui pré-requisitos, passo a passo com imagens da UI, exemplos de Connection URI, e um snippet Python usando
BaseHook+mysql.connector.
¶ 1) Pré-requisitos
- Airflow em execução (Webserver / Scheduler / Worker).
- Providers e driver Python instalados no(s) container(es) do Airflow:
apache-airflow-providers-mysqlmysql-connector-python
- Se você usa o
docker-compose.yamloficial, adicione na seçãoenvironmentde webserver, scheduler e workers:
environment:
- _PIP_ADDITIONAL_REQUIREMENTS=apache-airflow-providers-mysql mysql-connector-python
Ps. Provavelmente toda essa parte já vai vir configurada
¶ 2) Abrir a tela de Connections
No Airflow UI, vá em Admin → Connections. Veja imagem a seguir:
¶ 3) Criar uma nova conexão MySQL
Clique em Add Connection. Conforme imagem abaixo:
Preencha os campos:
- Connection Id: um identificador claro (ex.:
glpi_prod,clone_prod_atual). - Connection Type:
mysql(aparece após instalar o provider). - Host: IP/DNS do banco (ex.:
172.16.0.190). - Schema: nome do banco (ex.:
glpi). - Login / Password: credenciais do usuário do MySQL.
- Port:
3306(ou3307se você expôs nessa porta).
Salve a configuração.
💡 Dica: mantenha um padrão de nomes (ex.:
sistema_ambiente→glpi_prod,glpi_clone).
¶ 4) Conferir conexões criadas
Você verá sua lista de conexões cadastradas, como no exemplo abaixo:
¶ 5) Alternativa: definir via Connection URI (env var)
Você também pode declarar a conexão por variável de ambiente (útil em produção/CI):
export AIRFLOW_CONN_GLPI_PROD='mysql+mysqlconnector://USUARIO:SENHA@172.16.0.190:3306/GLPI?charset=utf8mb4'
Ps. Opcional
¶ 6) 6) Consumir a conexão no código (BaseHook + mysql.connector)
Imports Função de fábrica de conexão
import mysql.connector
from airflow.hooks.base import BaseHook
CONN_ID = "glpi_prod" # ou o ID que você cadastrou
def get_mysql_connection():
conn = BaseHook.get_connection(CONN_ID)
return mysql.connector.connect(
host=conn.host,
port=conn.port,
user=conn.login,
password=conn.password,
database=conn.schema,
autocommit=False, # ligue só se fizer sentido
)
¶ DAG de teste rápido
from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime
from contextlib import closing
def ping_mysql():
with closing(get_mysql_connection()) as db, db.cursor(dictionary=True) as cur:
cur.execute("SELECT 1 AS ok;")
print(cur.fetchone())
with DAG(
dag_id="teste_mysql_conn",
start_date=datetime(2024, 1, 1),
schedule="@once",
catchup=False,
) as dag:
PythonOperator(task_id="ping_mysql", python_callable=ping_mysql)
💡 Dica: Junte os dois códigos e mantenha as importações. A
def ping_mysql()deve vir após adef get_mysql_connection()
Rode um Trigger na DAG e verifique os logs. Se aparecer {‘ok’: 1}, a conexão está OK.
¶ 7) Boas práticas
- Usuário dedicado no MySQL (evite root).
- Permissões mínimas necessárias (ex.: apenas SELECT para rotinas de leitura).
- Ambientes separados: _dev, _hom, _prod.
- Não exponha senhas em código. Prefira Connections do Airflow ou Secrets Backends.
- Se necessário, use SSL (campo Extra da Connection) e charset=utf8mb4.
¶ 8) Solução de problemas (checklist)
⚠️ Aviso: “Connection Type missing” ao adicionar connection
→ Falta provider. Confirmeapache-airflow-providers-mysqlnos containers.
‼️🚨‼️ Erro: “Connection Type missing” ao adicionar connection
→ Falta provider. Confirmeapache-airflow-providers-mysqlnos containers.
¶ Erro de rede / timeout
→ Teste de dentro do container:
docker compose exec airflow-worker bash
apt-get update && apt-get install -y mysql-client || true
mysql -h 172.16.0.190 -P 3306 -u usuario -p -e "SELECT 1"
¶ Acesso negado
→ Verifique grants e host do usuário no MySQL:
CREATE USER 'airflow'@'%' IDENTIFIED BY 'SENHA_FORTE';
GRANT SELECT ON glpi.* TO 'airflow'@'%';
FLUSH PRIVILEGES;
¶ Porta incorreta
→ Garanta que a porta exposta (ex.: 3307) é a mesma configurada no Airflow.
¶ MySQL não aceita conexões remotas
→ Ajuste bind-address (0.0.0.0) e firewall/regra de rede conforme sua política.
¶ 9) (Opcional) Usando MySqlHook
Se preferir um hook do provider:
from airflow.providers.mysql.hooks.mysql import MySqlHook
def fetch_rows():
hook = MySqlHook(mysql_conn_id="glpi_prod")
rows = hook.get_records("SELECT COUNT(*) FROM users;")
print(rows)
¶ 10) Segurança em produção
Rotacione senhas e use Secret Backends (AWS Secrets Manager, HashiCorp Vault, etc.).
Limite o acesso por IP/rede (security groups, firewall).
Audite quem pode editar Connections na UI do Airflow.
Referências visuais (UI)
1. Admin → Connections
2. Add Connection
3. Connections criadas