Construindo uma Pipeline com Prefect 3

Guia completo para criar pipelines de dados usando Prefect 3 na Prefeitura do Rio de Janeiro

Construindo uma Pipeline com Prefect 3

Este guia explica como criar e configurar pipelines de dados usando Prefect 3 na Prefeitura do Rio de Janeiro, seguindo as melhores práticas e padrões estabelecidos pela equipe IplanRio.

✅ Pré-requisitos

Acessos e Permissões

• Acesso ao Tailscale para conexão à rede interna da Prefeitura
• Permissão para acessar o Infisical (gerenciador de secrets)
• Acesso de leitura ao projeto BigQuery rj-iplanrio
• Permissões de colaborador no repositório GitHub

Caso não tenha acesso ao GitHub, BigQuery ou Infisical, solicite permissões no canal #peça-permissão do Discord da IplanRio.

Ambiente de Desenvolvimento

• Python 3.13+
• uv package manager
• Editor de código (VSCode recomendado)
• WSL 2 (usuários Windows)
• Conhecimento básico de Git e GitHub

🔧 Configuração Inicial

1. Clonar o Repositório

git clone https://github.com/prefeitura-rio/prefect_rj_iplanrio.git
cd prefect_rj_iplanrio

2. Instalar Nix

O repositório utiliza Nix para gerenciar o ambiente de desenvolvimento.

curl --proto '=https' --tlsv1.2 -L https://nixos.org/nix/install | sh -s -- --daemon

Verifique a instalação:

which nix

3. Configurar direnv

O direnv gerencia automaticamente variáveis de ambiente do repositório.

curl -sfL https://direnv.net/install.sh | bash

Adicione o hook ao seu shell:

# Para Bash
eval "$(direnv hook bash)"

# Para Zsh, Oh My Zsh e outros shells, consulte: https://direnv.net/docs/hook.html

4. Instalar dependências

uv sync

5. Configurar pre-commit hooks

uv run pre-commit install --install-hooks

Pre-commit hooks garantem qualidade do código antes do commit. Veja mais em pre-commit.com.

🚀 Boas Práticas de Desenvolvimento

1. Estrutura de Branch

# Criar branch para sua pipeline
git checkout -b staging/nome-da-pipeline

# Exemplo para pipeline de saúde
git checkout -b staging/pipeline-dados-saude-sms

IMPORTANTE: Use sempre o prefixo staging/ no nome da branch para que o CI/CD reconheça e processe sua pipeline automaticamente.

2. Nomenclatura de Pipelines

Siga o padrão estabelecido:

# ✅ Correto: Nome descritivo e em snake_case
@flow(log_prints=True)
def rj_cvl__os_info():
    pass

# ❌ Evite: Nomes genéricos
@flow(log_prints=True)
def extracao_de_dados():
    pass

🧪 Criação de Nova Pipeline

1. Gerar Template com Cookiecutter

O repositório usa cookiecutter para criar pipelines padronizadas com Dockerfile, flow.py, prefect.yaml e pyproject.toml.

uvx cookiecutter templates --output-dir=pipelines

O comando solicitará informações sobre secretaria e pipeline:

$ uvx cookiecutter templates --output-dir=pipelines
secretaria [sms]: sms
pipeline [dados_pacientes]: dados_saude_sms

Use nomes descritivos que identifiquem a secretaria e o tipo de dados. Exemplo: censo_escolar (SME), pacientes_upa (SMS), multas_transito (SMTR).

2. Configurar flow.py

Edite flow.py com as configurações específicas da sua pipeline:

# Exemplo: Pipeline de dados de pacientes SMS
@flow(log_prints=True)
def rj_sms__dados_pacientes_upa():
    # Configurações da base de dados
    db_database = "saude_producao"
    db_host = "10.20.30.40"
    db_port = 3306
    db_type = "mysql"
    
    # Dataset de destino no BigQuery
    dataset_id = "saude_sms"
    
    # Caminho dos secrets no Infisical
    infisical_secret_path = "/db-saude-sms"

Credenciais Obrigatórias: Solicite ao IplanRio para adicionar usuário e senha do banco de origem no Infisical no caminho especificado em infisical_secret_path. Sem essas credenciais, a pipeline não conseguirá se conectar ao banco de dados.

Parâmetros de Configuração:

Parâmetro	Descrição	Exemplo
db_database	Nome do banco de dados de origem	banco1
db_host	Host do banco de dados de origem	10.15.255.1
db_port	Porta de conexão	3306
db_type	Tipo do banco de dados	mysql, postgres
dataset_id	Dataset de destino no BigQuery (projeto rj-iplanrio)	saude_sms
infisical_secret_path	Caminho dos secrets no Infisical	/db-saude-sms

3. Configurar prefect.yaml

Configure os schedules no arquivo prefect.yaml conforme suas necessidades:

Schedule Overwrite

Schedule overwrite substitui todos os dados da tabela a cada execução. Use para dados que devem refletir sempre o estado atual completo.

schedules:
  # Exemplo de schedule overwrite
  - interval: 86400 # executa a cada 24h
    anchor_date: "2022-01-01T01:00:00" # Data de inicio do schedule em UTC
    timezone: America/Sao_Paulo
    slug: dados_saude_sms_diario # slug do schedule
    parameters:
      table_id: pacientes_sms # Nome da tabela no BigQuery
      execute_query: |
        SELECT
          id_paciente,
          nome,
          data_nascimento,
          telefone
        FROM banco1.saude

Use overwrite para tabelas dimensão ou cadastros que precisam refletir o estado atual completo, sem dados históricos.

Schedule Incremental

Schedule incremental adiciona apenas dados novos ou atualizados. Use para tabelas fato ou registros transacionais com alto volume.

schedules:
  # Exemplo de schedule incremental
  - interval: 86400
    anchor_date: "2022-01-01T02:32:00"
    timezone: America/Sao_Paulo
    slug: dados_saude_sms_incremental
    parameters:
      table_id: pacientes_sms_incremental
      dump_mode: append
      partition_columns: data_atualizacao # Coluna de particionamento
      partition_date_format: "%Y-%m-%d" # Formato da data de particionamento
      break_query_frequency: day # Frequencia de particionamento
      break_query_start: current_day # Data de inicio do particionamento
      break_query_end: current_day # Data de fim do particionamento
      execute_query: |
        SELECT
          id_paciente,
          nome,
          data_nascimento,
          telefone,
          data_atualizacao
        FROM banco1.saude

Use incremental para dados que crescem continuamente (transações, eventos, logs). O particionamento otimiza queries e reduz custos no BigQuery.

Parâmetros de Schedule:

Parâmetro	Descrição	Valores
interval	Intervalo entre execuções em segundos	86400 (24h)
anchor_date	Data de início do schedule em UTC	2022-01-01T01:00:00
timezone	Fuso horário	America/Sao_Paulo
slug	Identificador único do schedule	dados_saude_sms_diario
table_id	Nome da tabela no BigQuery	pacientes_sms
dump_mode	Modo de inserção	append, overwrite
partition_columns	Coluna(s) para particionamento	data_atualizacao
partition_date_format	Formato da data de particionamento	%Y-%m-%d
break_query_frequency	Frequência de particionamento	day, month, year
break_query_start/end	Período de processamento	current_day

4. Commit e Push

# Adicionar arquivos
git add pipelines/sua_secretaria/sua_pipeline/

# Commit com mensagem descritiva
git commit -m "feat: adiciona pipeline de dados de saúde SMS

- Cria pipeline para extração de dados SMS
- Configura schedule diário incremental
- Adiciona particionamento por data"

# Push para branch
git push origin staging/sua-pipeline

Use Conventional Commits para mensagens padronizadas: feat:, fix:, docs:, refactor:.

5. Criar Pull Request

1. Criar PR no GitHub para a branch staging/sua-pipeline
2. Descrever mudanças: Explique o objetivo da pipeline e impacto no projeto
3. Solicitar review da equipe IplanRio
4. Aguardar CI/CD: Todos os testes devem passar
5. Testar em Staging: Valide a pipeline no ambiente de desenvolvimento após deploy
6. Merge: Após aprovação, faça merge para main

Workflow de CI/CD Automático

O repositório utiliza GitHub Actions para automatizar build, deploy e publicação de pipelines.

🚀 Deploy Automático

O sistema possui workflows separados para staging e produção:

Ambiente	Workflow	Trigger	Monitora
Staging	deploy-prefect-flows-staging.yaml	Push em staging/*	pipelines/**
Produção	deploy-prefect-flows-prod.yaml	Push em master	pipelines/**

🔧 Processo de Deploy

Ambos os workflows executam:

1. Checkout do código-fonte
2. Login no GitHub Container Registry (ghcr.io)
3. Instalação de dependências Python com uv
4. Execução do script .github/scripts/deploy_prefect_flows.py
   • Deploy automático de todos os flows em pipelines/*/prefect.yaml
   • Falhas interrompem o workflow e registram erro nos logs

🐳 Build da Imagem Base

Workflow build-and-push-root-dockerfile.yaml:

• Trigger: Alterações no Dockerfile raiz ou push em master
• Processo: Build e publicação em ghcr.io/${{ github.repository }}:latest

Staging permite testar pipelines antes de produção. Após deploy em staging funcionar, teste no ambiente de desenvolvimento. Apenas após merge em master as pipelines são deployadas em produção.

Se algum deploy falhar, o workflow será interrompido. Corrija os problemas antes de tentar novamente.

📊 Monitoramento

• Acompanhe progresso na aba Actions do GitHub
• Verifique logs para identificar erros
• Aguarde conclusão antes de solicitar review
• Falhas requerem novo commit para re-executar CI/CD

🔧 Troubleshooting

Qual work-pool utilizar:

A escolha do work-pool depende de onde a pipeline será executada e dos recursos que ela precisa acessar.

Work-pool	Quando utilizar
default-pool	Utilize sempre que a pipeline não precisar acessar recursos internos da Iplanrio e não pertencer ao projeto Táxi.Rio.
k3s-pool	Pool padrão para a maioria das pipelines. Utilize para pipelines que precisam acessar recursos internos da Iplanrio, como bancos de dados, APIs, serviços ou outros recursos disponíveis apenas na infraestrutura interna.
taxirio-pool	Utilize exclusivamente para pipelines do projeto Táxi.Rio.

Erro de Conexão com Banco de Dados:

• Verifique credenciais no Infisical no caminho infisical_secret_path
• Confirme acesso ao host e porta via Tailscale
• Teste conexão manualmente com ferramenta como mysql-client ou psql

Falha no Deploy:

• Verifique logs na aba Actions do GitHub
• Confirme que todos os arquivos foram commitados (flow.py, prefect.yaml, Dockerfile)
• Valide sintaxe YAML em yamllint.com

Pipeline Não Executa no Schedule:

• Verifique se anchor_date está no passado (não no futuro)
• Confirme timezone: America/Sao_Paulo
• Valide interval em segundos (86400 = 24 horas)

Dados Não Aparecem no BigQuery:

• Confirme dataset_id correto no projeto rj-iplanrio
• Verifique se a query retorna dados executando-a manualmente
• Valide permissões de escrita no BigQuery (solicite ao IplanRio se necessário)

📚 Recursos Adicionais

• Migração de Pipeline
• Repositório GitHub
• Documentação Prefect

---