Erathos + dbt Cloud: Como Orquestrar Pipelines ELT com Webhooks (Guia Completo)

Em projetos modernos de dados, a separação de responsabilidades entre ingestão e transformação se tornou uma prática fundamental. A Erathos resolve o problema da extração e carga de dados (EL) de diversas fontes para o data warehouse, enquanto o dbt se especializa na transformação (T) desses dados em modelos analíticos confiáveis.

Essa separação, no entanto, nos faz questionar: como orquestrar esses processos de forma eficiente? Como garantir que o dbt execute transformações apenas após a chegada de novos dados? Como fazer a Erathos reagir a falhas em transformações críticas do dbt?

A API de Orquestração da Erathos resolve esse problema ao permitir a criação de pipelines orientados a eventos. Com os webhooks, os fluxos são acionados automaticamente por acontecimentos reais como término de ingestões, falhas ou variações no volume de dados reduzindo a dependência de agendamentos fixos que nem sempre refletem o cenário atual.

A arquitetura que implementaremos neste tutorial representa um padrão moderno de engenharia de dados:

1. Erathos ingere dados transacionais de sistemas operacionais (PostgreSQL, MongoDB, APIs) para o BigQuery

2. Webhooks disparam automaticamente jobs do dbt Cloud quando novas cargas são concluídas

3. dbt Cloud transforma os dados brutos em modelos analíticos (facts e dimensions)

4. Orquestração reversa permite que o dbt também controle execuções na Erathos quando necessário

O Problema: Orquestração Desacoplada

Antes de implementar a solução, é importante entender os problemas que ela resolve:

Cenário 1: Agendamentos Fixos Desconectados

Situação: A Erathos está agendada para executar às 2h da manhã, e o dbt às 3h. Se a ingestão da Erathos falhar ou atrasar, o dbt processará dados desatualizados ou falhará por falta de dados novos.

Solução: O dbt executa somente quando a ingestão da Erathos termina com sucesso e após pelo menos 1 linha ter sido transferida.

Cenário 2: Falta de Rastreabilidade

Situação: Um dashboard apresenta dados incorretos. Não há clareza sobre qual etapa do pipeline falhou ou está desatualizada.

Solução: Cada execução do dbt recebe metadados da execução da Erathos (execution_id, número de linhas, tabelas afetadas), permitindo rastreamento completo.

Cenário 3: Desperdício de Recursos

Situação: O dbt executa periodicamente mesmo quando não há dados novos, consumindo créditos do BigQuery desnecessariamente.

Solução: Webhooks disparam apenas quando há dados novos (ROWS > 0), otimizando o uso de recursos.

Arquitetura da Solução

O fluxo implementado segue esta estrutura:

Conhecimento prévio

Este tutorial assume familiaridade com:

• Conceitos básicos de dbt (models, sources, refs)

• Navegação na interface do dbt

• Conceitos de REST APIs e webhooks

• Estrutura básica de data warehouses (facts e dimensions)

Pré-requisitos

Ambiente Erathos

• Workspace ativo

• Fonte de dados conectada (PostgreSQL, MongoDB, API, etc.)

• Job de ingestão configurado e testado

• Destino configurado para BigQuery

Ambiente dbt Cloud

• Projeto dbt conectado ao BigQuery

• Models já criados e testados

• Job configurado no dbt Cloud

BigQuery

• Projeto ativo no Google Cloud

• Datasets criados:

  • raw_data (dados brutos da Erathos)

  • staging (camada intermediária do dbt)

  • analytics (camada final - facts e dimensions)

Passo 0: Coletando Informações Necessárias

Antes de começar, você precisará coletar algumas informações. Use esta tabela para organizar:

Passo 1: Configurar Credenciais de Autenticação

A integração via API requer autenticação em ambas as plataformas. As credenciais serão armazenadas de forma segura usando o sistema de Variables e Secrets da Erathos.

Gerar API Key da Erathos

1. Acesse Settings > User no Erathos

2. Na seção API Key, clique em Create

3. IMPORTANTE: A chave é exibida apenas uma vez. Copie e armazene em local seguro

4. Opcionalmente, defina uma data de expiração

Exemplo de API Key:

Gerando Personal Token do dbt Cloud

1. Acesse sua conta no dbt Cloud

2. Navegue até Account Settings > API Access

3. Clique em Create Service Token (ou Personal Token)

4. Copie o token gerado

5. Copie também o Account ID (você vai precisar)

Exemplo de configuração:

Criar Variables e Secrets na Erathos

A Erathos permite centralizar valores usados com frequência (como URLs e credenciais) no formato chave → valor.

• Variables: para valores não sensíveis, que podem ser reutilizados em webhooks e outros processos.

• Secrets: para dados sensíveis, como credenciais, que não podem ser visualizados após a criação, garantindo mais segurança.

Criando a Variable dbt_account_id

Via API:

curl -X POST <https://api.erathos.com/developers/workspaces/{workspace_id}/variables/> \\
  -H "Authorization: Api-Key {sua_api_key_erathos}" \\
  -H "Content-Type: application/json" \\
  -d '{
    "name": "dbt_account_id",
    "value": "12345"
  }'

Criando o Secret dbt_personal_token

Via API:

curl -X POST <https://api.erathos.com/developers/workspaces/{workspace_id}/secrets/> \\
  -H "Authorization: Api-Key {sua_api_key_erathos}" \\
  -H "Content-Type: application/json" \\
  -d '{
    "name": "dbt_personal_token",
    "value": "dbt_a1b2c3d4e5f6..."
  }'

Passo 2: Identificar IDs Necessários

Antes de criar webhooks, precisamos coletar os identificadores únicos dos recursos que vamos orquestrar.

Obtendo Job ID da Erathos

Cada conexão de dados na Erathos é representada por um Job. O UUID deste job é necessário para configuração.

Via Interface:

1. Navegue até Jobs

2. Localize o job desejado (ex: orders vindo do PostgreSQL)

3. Copie o Job ID da URL ou da interface

Via API:

curl -X GET "<https://api.erathos.com/developers/workspaces/{workspace_id}/jobs/>" \\
  -H "Authorization: Api-Key {sua_api_key_erathos}"

Resposta (exemplo):

{
  "count": 3,
  "results": [
    {
      "_id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
      "name": "orders",
      "schema_name": "raw_data",
      "connector_name": "PostgreSQL",
      "is_active": true
    },
    {
      "_id": "b2c3d4e5-f6a7-8901-bcde-f12345678901",
      "name": "products",
      "schema_name": "raw_data",
      "connector_name": "PostgreSQL",
      "is_active": true
    }
  ]
}

Anote os IDs:

• Job orders: a1b2c3d4-e5f6-7890-abcd-ef1234567890

• Job products: b2c3d4e5-f6a7-8901-bcde-f12345678901

Obtendo Job ID do dbt Cloud

No dbt Cloud, cada execução agendada ou ad-hoc é configurada como um Job. Precisamos do ID numérico desse job.

Via Interface:

1. No dbt Cloud, navegue até Deploy > Jobs

2. Clique no job que você quer triggerar (ex: "Daily Analytics Refresh")

3. O Job ID estará na URL: https://cloud.getdbt.com/deploy/{account_id}/projects/{project_id}/jobs/{job_id}

Via API:

curl -X GET "<https://cloud.getdbt.com/api/v2/accounts/{account_id}/jobs/>" \\
  -H "Authorization: Bearer {dbt_personal_token}"

Exemplo de Job ID: 54321

Passo 3: Erathos Disparando dbt Cloud

Após a Erathos terminar de ingerir dados no BigQuery, dispara automaticamente um job no dbt Cloud para transformá-los.

Entendendo o Fluxo

Anatomia do Webhook

Um webhook no Erathos contém:

1. description: Descrição para identificar o webhook

2. is_active: Se está ativo ou não

3. jobs: Lista de Job IDs do Erathos que devem triggerar o webhook

4. method: Tipo de requisição HTTP (POST, GET, DELETE, etc.)

5. url: Endpoint de destino (dbt Cloud API)

6. header: Headers HTTP (autenticação, content-type)

7. body: Corpo da requisição (dados enviados para o dbt)

8. rules: Condições que devem ser satisfeitas para o webhook disparar

Criando o Webhook: Erathos → dbt Cloud

curl -X POST "<https://api.erathos.com/developers/workspaces/{workspace_id}/orchestration/webhooks/>" \\
  -H "Authorization: Api-Key {sua_api_key_erathos}" \\
  -H "Content-Type: application/json" \\
  -d '{
    "description": "Trigger dbt Cloud job after successful orders ingestion",
    "is_active": true,
    "method": "POST",
    "url": "<https://cloud.getdbt.com/api/v2/accounts/${{variables.dbt_account_id}>}/jobs/54321/run/",
    "header": {
      "Content-Type": "application/json",
      "Accept": "application/json",
      "Authorization": "Bearer ${{secrets.dbt_personal_token}}"
    },
    "body": {
      "cause": "Triggered by Erathos after successful ingestion of ${{erathos.TABLE_NAME}} | Execution ID: ${{erathos.EXECUTION_ID}} | Rows: ${{erathos.ROWS}}"
    },
    "rules": [
      {
        "variable_name": "STATUS",
        "operation": "EQUAL",
        "value": "FINISHED"
      },
      {
        "variable_name": "ROWS",
        "operation": "GREATER_THAN",
        "value": "0"
      },
      {
        "variable_name": "NESTED_TABLE",
        "operation": "EQUAL",
        "value": "false"
      }
    ],
    "jobs": [
      "a1b2c3d4-e5f6-7890-abcd-ef1234567890"
    ]
  }'

Detalhando os Componentes

URL Dinâmica

"url": "<https://cloud.getdbt.com/api/v2/accounts/${{variables.dbt_account_id}>}/jobs/54321/run/"

• ${{variables.dbt_account_id}}: Substituído automaticamente pelo valor da variable criada no Passo 1

• 54321: ID do job do dbt Cloud obtido no Passo 2

• /run/: Endpoint da API do dbt Cloud para iniciar execuções

Headers de Autenticação

"header": {
  "Authorization": "Bearer ${{secrets.dbt_personal_token}}"
}

O token é referenciado de forma segura através do sistema de Secrets.

Body com Metadados

"body": {
  "cause": "Triggered by Erathos after successful ingestion of ${{erathos.TABLE_NAME}}"
}

• ${{erathos.TABLE_NAME}}: Nome da tabela ingerida (ex: "orders")

• ${{erathos.EXECUTION_ID}}: UUID único da execução do Erathos

• ${{erathos.ROWS}}: Número de linhas transferidas

Resultado no dbt Cloud: Quando o job do dbt rodar, você verá na interface:

Esta informação facilita o troubleshooting, permitindo rastrear qual execução do Erathos disparou cada execução do dbt.

Rules: Condições Inteligentes

As rules determinam quando o webhook deve disparar.

"rules": [
  {
    "variable_name": "STATUS",
    "operation": "EQUAL",
    "value": "FINISHED"
  },
  {
    "variable_name": "ROWS",
    "operation": "GREATER_THAN",
    "value": "0"
  },
  {
    "variable_name": "NESTED_TABLE",
    "operation": "EQUAL",
    "value": "false"
  }
]

Traduzindo para lógica:

Por que essas regras?

1. STATUS == "FINISHED": Só dispara se a ingestão foi bem-sucedida (não dispara em caso de falha)

2. ROWS > 0: Evita processar execuções vazias (economiza créditos do BigQuery)

3. NESTED_TABLE == false: Para conectores de APIs que têm múltiplas tabelas aninhadas, garante que só a tabela principal dispare o webhook (evita múltiplos triggers)

Testando o Webhook

Após criar o webhook, teste manualmente:

1. Na Erathos: Execute manualmente o job orders

2. Acompanhe: A execução deve terminar com sucesso e transferir pelo menos 1 linha

3. Verifique no dbt Cloud: Navegue até Deploy > Run History

4. Valide: Você deve ver uma nova execução com a causa personalizada

Exemplo de log esperado:

Passo 4: Monitoramento e Troubleshooting

Visualizando Webhooks Criados

Para listar todos os webhooks configurados no workspace:

curl -X GET "<https://api.erathos.com/developers/workspaces/{workspace_id}/orchestration/webhooks/>" \\
  -H "Authorization: Api-Key {erathos_api_key}"

Resposta:

{
  "count": 2,
  "results": [
    {
      "_id": "webhook-uuid-123",
      "description": "Trigger dbt Cloud job after successful orders ingestion",
      "is_active": true,
      "method": "POST",
      "url": "<https://cloud.getdbt.com/api/v2/accounts/12345/jobs/54321/run/>",
      "jobs": ["a1b2c3d4-e5f6-7890-abcd-ef1234567890"]
    },
    {
      "_id": "webhook-uuid-456",
      "description": "Run job_b after job_a completes successfully",
      "is_active": true,
      "method": "POST",
      "jobs": ["{job_a_id}"]
    }
  ]
}

Desativando Temporariamente um Webhook

Se você precisa pausar temporariamente a orquestração (ex: durante manutenção do dbt Cloud):

curl -X PUT "<https://api.erathos.com/developers/workspaces/{workspace_id}/orchestration/webhooks/{webhook_id}/>" \\
  -H "Authorization: Api-Key {erathos_api_key}" \\
  -H "Content-Type: application/json" \\
  -d '{
    "is_active": false
  }'

Conclusão

Você implementou com sucesso uma arquitetura de orquestração entre Erathos e dbt Cloud, criando pipelines de dados verdadeiramente event-driven e inteligentes.

• Automação Declarativa: Webhooks que reagem automaticamente a eventos reais (conclusão de ingestões, volumes de dados, horários), eliminando a necessidade de orquestradores complexos.

• Rastreabilidade Completa: Metadados da Erathos fluem para o dbt Cloud, permitindo debugar falhas e entender o histórico de cada execução.

• Eficiência Operacional: Rules garantem que transformações só rodam quando realmente necessário, economizando créditos do BigQuery e tempo de processamento.

• Flexibilidade Arquitetural: A mesma API suporta integrações com Airflow, Prefect, sistemas customizados e orquestração interna da própria Erathos.

Referências

• Documentação Oficial da API Erathos: https://docs.erathos.com/api

• Documentação da API do dbt Cloud: https://docs.getdbt.com/dbt-cloud/api-v2

• Erathos - Integração com dbt Cloud: https://docs.erathos.com/orchestration/dbt-cloud

• Erathos - Variables e Secrets: https://docs.erathos.com/platform/variables-and-secrets

• dbt Cloud - Webhooks (Outbound): https://docs.getdbt.com/docs/deploy/webhooks