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

Acesso de leitura ao projeto BigQuery rj-iplanrio
Permissões de colaborador no repositório GitHub
Conhecimento básico de Git e GitHub
Python 3.8+ instalado
uv package manager instalado

🔧 Configuração Inicial

1. Clonar o Repositório

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

🚀 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. Usar Template Cookiecutter

O repositório utiliza cookiecutter para facilitar a criação de novas pipelines de forma padronizada. Os templates disponíveis em templates/ permitem gerar rapidamente a estrutura de diretórios e arquivos necessários para uma nova pipeline Prefect, incluindo Dockerfile, flow.py, prefect.yaml e pyproject.toml. Para criar uma nova pipeline, instale uv e rode:

uvx cookiecutter templates --output-dir=pipelines

Você será solicitado a informar valores como secretaria e pipeline, que serão utilizados para preencher os nomes dos diretórios, arquivos e variáveis nos templates. O template gerado já segue o padrão de nomenclatura definido anteriormente. Exemplo de interação:

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

2. Configurar flow.py

Após gerar o template, substitua os valores padrão no arquivo flow.py pelos valores específicos da sua pipeline:

IMPORTANTE: É necessário solicitar à equipe de desenvolvimento da IplanRio para adicionar o usuário e senha do seu banco de origem no Infisical no caminho especificado em infisical_secret_path. Sem essas credenciais configuradas, sua pipeline não conseguirá se conectar ao banco de dados.

# Exemplo de configuração em flow.py
@flow(log_prints=True)
def pipeline_dados_saude_sms():
    # Configurações da base de dados
    db_database = "banco1"  # Nome do Banco de Origem
    db_host = "10.15.255.1"  # Host do Banco de Origem
    db_port = 3306  # Porta do Banco de Origem
    db_type = "mysql"  # Tipo do Banco de Origem
    
    # Configurações do dataset
    dataset_id = "nome_do_dataset"  # Dataset de destino em rj-iplanrio
    
    # Configurações do Infisical
    infisical_secret_path = "/db-seu-banco-de-origem"  # Caminho dos secrets
    

Parâmetros importantes:

db_database: Nome do banco de dados de origem (ex: banco1)
db_host: Host do banco de dados de origem (ex: 10.15.255.1)
db_port: Porta de conexão do banco de dados de origem (ex: 3306)
db_type: Tipo do banco de dados de origem (ex: mysql, postgres, etc)
dataset_id: Dataset de destino no BigQuery (todas as extrações são adicionadas ao projeto rj-iplanrio)
infisical_secret_path: Caminho dos secrets no Infisical

3. Configurar prefect.yaml

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

Schedule Overwrite

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

Schedule Incremental (Incremental)

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

Explicação dos parâmetros:

interval: Intervalo entre as runs em segundos (86400 = 24 horas)
anchor_date: Data de início do schedule em UTC
timezone: Fuso horário (America/Sao_Paulo)
slug: Identificador único do schedule
table_id: Nome da tabela no BigQuery
dump_mode: Modo de inserção (append para incremental, overwrite para substituição)
partition_columns: Coluna(s) para particionamento
partition_date_format: Formato da data de particionamento
break_query_frequency: Frequência de particionamento (day, month, year)
break_query_start/end: Período de dados a serem processados

4. Commit e Push

# 1. Adicionar arquivos
git add pipelines/sua_secretaria/sua_pipeline/

# 2. 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"

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

5. Pull Request

Criar PR no GitHub
Descrição: Explicar mudanças e impacto
Review: Solicitar revisão da equipe IplanRio
Testes: Garantir que todos os testes passem
Teste em Staging: Após o deploy em staging funcionar, você já pode testar sua pipeline no ambiente de desenvolvimento
Merge: Após aprovação, merge para main

Workflow de CI/CD Automático

O repositório utiliza GitHub Actions para automatizar todo o ciclo de vida das pipelines, incluindo build, deploy e publicação de imagens Docker: 🚀 Deploy Automático dos Flows: O sistema possui dois workflows principais para deploy automático:

.github/workflows/deploy-prefect-flows-staging.yaml - Ambiente de Staging
- Acionado a cada push em branches staging/*
- Pode ser executado manualmente
- Monitora alterações em pipelines/**
.github/workflows/deploy-prefect-flows-prod.yaml - Ambiente de Produção
- Acionado a cada push na branch master
- Pode ser executado manualmente
- Monitora alterações em pipelines/**

🔧 Processo de Deploy: Ambos os workflows executam as seguintes etapas:

Checkout do código-fonte
Login no GitHub Container Registry (ghcr.io)
Instalação das dependências Python com uv
Execução do script de deploy .github/scripts/deploy_prefect_flows.py
- Faz o deploy automático de todos os flows definidos em pipelines/*/prefect.yaml
- Caso algum deploy falhe, o workflow é interrompido e o erro é registrado nos logs

🐳 Build e Publicação da Imagem Base: O workflow .github/workflows/build-and-push-root-dockerfile.yaml:

Acionado: Em alterações no Dockerfile da raiz ou push na branch master
Execução manual: Disponível quando necessário
Processo:
- Build da imagem Docker definida no Dockerfile do repositório
- Publicação da imagem no GitHub Container Registry (ghcr.io/${{ github.repository }}:latest)

O workflow de staging permite testar suas pipelines em um ambiente isolado antes do deploy em produção. Após o deploy em staging funcionar, você já pode testar sua pipeline no ambiente de desenvolvimento. Apenas após o merge na branch master é que as pipelines são deployadas em produção.

IMPORTANTE: Se algum deploy falhar durante o processo, o workflow será interrompido e você precisará corrigir os problemas antes de tentar novamente.

📊 Monitoramento do CI/CD:

Acompanhe o progresso dos deploys na aba “Actions” do seu repositório
Verifique os logs para identificar possíveis erros
Aguarde a conclusão bem-sucedida antes de solicitar review da equipe
Se houver falhas, corrija os problemas e faça novo commit - o CI será executado novamente

🔧 Troubleshooting

Problemas Comuns

Erro de conexão com banco de dados:

Verifique se as credenciais estão corretas no Infisical
Confirme se o host e porta estão acessíveis
Teste a conexão manualmente

Falha no deploy:

Verifique os logs na aba “Actions” do GitHub
Confirme se todos os arquivos necessários foram commitados
Valide a sintaxe do prefect.yaml

Pipeline não executa no schedule:

Verifique se o anchor_date está no passado
Confirme se o timezone está correto
Valide se o interval está em segundos

Dados não aparecem no BigQuery:

Verifique se o dataset_id está correto
Confirme se a query está retornando dados
Valide as permissões de escrita no projeto rj-iplanrio

Visão Geral

Guia de Desenvolvimento

Construindo uma Pipeline com Prefect 3

Construindo uma Pipeline com Prefect 3

✅ Pré-requisitos

🔧 Configuração Inicial

1. Clonar o Repositório

🚀 Boas Práticas de Desenvolvimento

1. Estrutura de Branch

2. Nomenclatura de Pipelines

🧪 Criação de Nova Pipeline

1. Usar Template Cookiecutter

2. Configurar flow.py

3. Configurar prefect.yaml

Schedule Overwrite

Schedule Incremental (Incremental)

4. Commit e Push

5. Pull Request

Workflow de CI/CD Automático

🔧 Troubleshooting

Problemas Comuns

📚 Recursos Adicionais

Visão Geral

Guia de Desenvolvimento

​Construindo uma Pipeline com Prefect 3

​✅ Pré-requisitos

​🔧 Configuração Inicial

​1. Clonar o Repositório

​🚀 Boas Práticas de Desenvolvimento

​1. Estrutura de Branch

​2. Nomenclatura de Pipelines

​🧪 Criação de Nova Pipeline

​1. Usar Template Cookiecutter

​2. Configurar flow.py

​3. Configurar prefect.yaml

​Schedule Overwrite

​Schedule Incremental (Incremental)

​4. Commit e Push

​5. Pull Request

​Workflow de CI/CD Automático

​🔧 Troubleshooting

​Problemas Comuns

​📚 Recursos Adicionais

Construindo uma Pipeline com Prefect 3

✅ Pré-requisitos

🔧 Configuração Inicial

1. Clonar o Repositório

🚀 Boas Práticas de Desenvolvimento

1. Estrutura de Branch

2. Nomenclatura de Pipelines

🧪 Criação de Nova Pipeline

1. Usar Template Cookiecutter

2. Configurar flow.py

3. Configurar prefect.yaml

Schedule Overwrite

Schedule Incremental (Incremental)

4. Commit e Push

5. Pull Request

Workflow de CI/CD Automático

🔧 Troubleshooting

Problemas Comuns

📚 Recursos Adicionais