mongodb¶

fonte

lê documentos de uma collection mongodb e os transforma em registros tabulares para processamento downstream.

quando usar¶

extrair logs, eventos ou dados de sessão armazenados em coleções mongodb
sincronizar documentos de um catálogo de produto (ou outro dado de schema flexível) para um lakehouse analítico
transformar dados semiestruturados (documentos com campos variáveis) em tabelas estruturadas para bi e relatórios

configuração¶

campo	tipo	obrigatório	descrição
`credential_id`	uuid	sim	conexão mongodb criada em configurações → conexões
`database`	string	não	nome do banco · padrão: schema configurado na conexão
`collection`	string	sim	nome da collection
`filter`	object	não	filtro mongodb (sintaxe mql padrão) · padrão: `{}`
`projection`	object	não	campos a incluir/excluir · padrão: todos os campos
`limit`	number	não	máximo de documentos a retornar · `0` = sem limite

extração incremental (watermark)¶

qualquer valor do filter aceita {{ pipeline.watermark }}, incluindo dentro de operadores mql ($gt, $gte etc.) — a interpolação percorre o documento inteiro, não só o nível superior:

{
  "filter": { "atualizado_em": { "$gt": "{{ pipeline.watermark }}" } },
  "watermark_column": "atualizado_em",
  "watermark_var": "watermark"
}

não é cdc

a extração incremental é por polling mql — não é o change stream nativo do mongodb. cobre bem o caso de coleções com um campo de última atualização confiável.

exemplos¶

leitura completa de uma collection¶

{
  "credential_id": "uuid-da-conexao",
  "database": "app_producao",
  "collection": "pedidos"
}

com filtro por data¶

{
  "credential_id": "uuid-da-conexao",
  "database": "app_producao",
  "collection": "pedidos",
  "filter": {
    "created_at": { "$gte": "2024-01-01T00:00:00Z" },
    "status": "finalizado"
  }
}

com projeção (selecionar campos)¶

{
  "credential_id": "uuid-da-conexao",
  "database": "app_producao",
  "collection": "usuarios",
  "projection": { "_id": 0, "email": 1, "nome": 1, "created_at": 1, "plano": 1 },
  "limit": 10000
}

filtrando por operador in¶

{
  "credential_id": "uuid-da-conexao",
  "database": "app_producao",
  "collection": "eventos",
  "filter": { "tipo": { "$in": ["compra", "cancelamento", "reembolso"] } }
}

estrutura de dados¶

documentos mongodb são convertidos para linhas tabulares — o campo _id é convertido para string:

documento mongodb:
{
  "_id": ObjectId("..."),
  "nome": "João",
  "endereco": { "cidade": "SP", "estado": "SP" },
  "tags": ["premium", "ativo"]
}

↓ após conversão

| _id         | nome  | endereco          | tags                 |
|-------------|-------|--------------------|-----------------------|
| "507f1f..." | João  | {"cidade": "SP"}  | ["premium","ativo"]  |

campos aninhados (objeto/array) chegam como texto json na coluna. use os nós expandir json, expandir struct ou explodir array para desestruturá-los em colunas próprias.

documentos com schema variável

quando os documentos de uma collection têm campos diferentes entre si, o schema resultante é a união de todos os campos vistos na extração. colunas ausentes em alguns documentos recebem valor nulo.

configurando a conexão¶

em configurações → conexões, crie uma conexão do tipo mongodb com host, porta e, se necessário, credencial de usuário/senha (opcional para mongo sem autenticação).