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

Criar PR no GitHub para a branch staging/sua-pipeline
Descrever mudanças: Explique o objetivo da pipeline e impacto no projeto
Solicitar review da equipe IplanRio
Aguardar CI/CD: Todos os testes devem passar
Testar em Staging: Valide a pipeline no ambiente de desenvolvimento após deploy
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:

Checkout do código-fonte
Login no GitHub Container Registry (ghcr.io)
Instalação de dependências Python com uv
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

Visão Geral

Guia de Desenvolvimento

Acesso aos Dados

Construindo uma Pipeline com Prefect 3

Construindo uma Pipeline com Prefect 3

✅ Pré-requisitos

Acessos e Permissões

Ambiente de Desenvolvimento

🔧 Configuração Inicial

1. Clonar o Repositório

2. Instalar Nix

3. Configurar direnv

4. Instalar dependências

5. Configurar pre-commit hooks

🚀 Boas Práticas de Desenvolvimento

1. Estrutura de Branch

2. Nomenclatura de Pipelines

🧪 Criação de Nova Pipeline

1. Gerar Template com Cookiecutter

2. Configurar flow.py

3. Configurar prefect.yaml

Schedule Overwrite

Schedule Incremental

4. Commit e Push

5. Criar Pull Request

Workflow de CI/CD Automático

🔧 Troubleshooting

Qual work-pool utilizar:

Erro de Conexão com Banco de Dados:

Falha no Deploy:

Pipeline Não Executa no Schedule:

Dados Não Aparecem no BigQuery:

📚 Recursos Adicionais

​Construindo uma Pipeline com Prefect 3

​✅ Pré-requisitos

​Acessos e Permissões

​Ambiente de Desenvolvimento

​🔧 Configuração Inicial

​1. Clonar o Repositório

​2. Instalar Nix

​3. Configurar direnv

​4. Instalar dependências

​5. Configurar pre-commit hooks

​🚀 Boas Práticas de Desenvolvimento

​1. Estrutura de Branch

​2. Nomenclatura de Pipelines

​🧪 Criação de Nova Pipeline

​1. Gerar Template com Cookiecutter

​2. Configurar flow.py

​3. Configurar prefect.yaml

​Schedule Overwrite

​Schedule Incremental

​4. Commit e Push

​5. Criar Pull Request

​Workflow de CI/CD Automático

​🔧 Troubleshooting

​Qual work-pool utilizar:

​Erro de Conexão com Banco de Dados:

​Falha no Deploy:

​Pipeline Não Executa no Schedule:

​Dados Não Aparecem no BigQuery:

​📚 Recursos Adicionais

Construindo uma Pipeline com Prefect 3

✅ Pré-requisitos

Acessos e Permissões

Ambiente de Desenvolvimento

🔧 Configuração Inicial

1. Clonar o Repositório

2. Instalar Nix

3. Configurar direnv

4. Instalar dependências

5. Configurar pre-commit hooks

🚀 Boas Práticas de Desenvolvimento

1. Estrutura de Branch

2. Nomenclatura de Pipelines

🧪 Criação de Nova Pipeline

1. Gerar Template com Cookiecutter

2. Configurar flow.py

3. Configurar prefect.yaml

Schedule Overwrite

Schedule Incremental

4. Commit e Push

5. Criar Pull Request

Workflow de CI/CD Automático

🔧 Troubleshooting

Qual work-pool utilizar:

Erro de Conexão com Banco de Dados:

Falha no Deploy:

Pipeline Não Executa no Schedule:

Dados Não Aparecem no BigQuery:

📚 Recursos Adicionais