Flow

Um flow é um pipeline de ETL. Ele vive em flows/<slug>--<id>.yaml, extrai dados de uma origem, transforma e grava numa tabela do DW.

Crie o arquivo com lumo new flow --name "Fato Vendas", edite, valide com lumo lint e publique com lumo push flow:<id>.

Ligue o autocomplete e a validação no seu editor colando esta linha no topo do arquivo:

yaml

# yaml-language-server: $schema=https://docs.horusbi.com.br/schemas/v2/flow.schema.json

Modelo de configuração

O arquivo tem dois documentos YAML separados por ---. O primeiro é o header, imutável depois da criação. O segundo é o body, que é o que você edita.

yaml

# ── header ────────────────────────────────────────────────
id: integer                       # obrigatório, >= 1, atribuído pelo servidor
kind: flow                        # obrigatório, literal "flow"
lumo: v2                          # obrigatório, literal "v2"
tenantId: integer                 # obrigatório, >= 1
---
# ── body ──────────────────────────────────────────────────
nome: string                      # obrigatório, mínimo 1 caractere
load_type: string                 # obrigatório: Total | Incremental | Temporal
load_type_column: string | null   # obrigatório quando load_type é Incremental ou Temporal

key_sweep:                        # opcional, só em Incremental com destino de chave única
  sql: string                     # SELECT devolvendo SÓ as colunas de chave, sem filtro
  credentialId: integer           # credencial da consulta, à parte da extração

tokenId: integer | null           # agente que executa o flow; null = agente padrão do tenant
development_variables: []         # variáveis de entrada por execução
tags: [string]

nodes:                            # obrigatório
  Processors:                     # obrigatório, lista de nós
    - internalId: string          # obrigatório, único dentro do flow
      kind: string                # obrigatório: ExtractPostgreSQL | Join | InsertDatawarehouse | ...
      description: string | null
      inputs: [string]            # internalId (ou complete_id) dos nós de entrada; a ordem importa
      options: {}                 # parâmetros do nó; as chaves dependem de kind
      metadata: {}
      complete_id: string         # gerado pelo servidor
      x: number                   # posição no editor visual
      y: number
  Connections:                    # obrigatório, arestas do DAG
    - from: string                # obrigatório, internalId de origem
      to: string                  # obrigatório, internalId de destino

table:                            # tabela do DW alimentada por este flow
  id: integer
  nome: string
owners:
  - userId: integer

nodes declara a mesma aresta em dois lugares: inputs no nó de destino e um item em Connections. Os dois precisam concordar. O lumo lint recusa um inputs que aponte para um nó inexistente.

Configuração completa

Um fato de vendas que junta o cabeçalho do pedido com os itens e grava no DW.

yaml

# flows/fato-vendas--44531.yaml
# yaml-language-server: $schema=https://docs.horusbi.com.br/schemas/v2/flow.schema.json
id: 44531
kind: flow
lumo: v2
tenantId: 853
---
nome: Fato Vendas
load_type: Temporal
# Temporal exige load_type_column, e a coluna precisa ser IMUTÁVEL.
# Use a data do fato (emissão), nunca uma data que o ERP reescreve (atualização).
load_type_column: DATA_EMISSAO
tokenId: 508
development_variables: []

nodes:
  Processors:
    - internalId: ext_pedidos
      kind: ExtractPostgreSQL
      description: Cabeçalho dos pedidos
      inputs: []
      options:
        # Chave da credencial, não o id. Descubra com: lumo list credential
        Credential: erp-producao
        SQL: |
          SELECT
            p.id            AS PEDIDO_ID,
            p.cliente_id    AS CLIENTE_ID,
            p.data_emissao  AS DATA_EMISSAO,
            p.valor_total   AS VALOR_TOTAL
          FROM public.pedidos p
          WHERE p.data_emissao >= '{StartDate}'
            AND p.data_emissao <= '{EndDate}'
      x: 100
      y: 200

    - internalId: ext_itens
      kind: ExtractPostgreSQL
      description: Itens do pedido
      inputs: []
      options:
        Credential: erp-producao
        SQL: |
          SELECT
            i.pedido_id   AS PEDIDO_ID,
            i.produto_id  AS PRODUTO_ID,
            i.quantidade  AS QUANTIDADE,
            i.valor_item  AS VALOR_ITEM
          FROM public.pedido_itens i
      x: 100
      y: 400

    - internalId: join_itens
      kind: Join
      description: Enriquece cada item com o cabeçalho
      # A ordem importa: inputs[0] é o primário (todas as colunas passam),
      # inputs[1] é o secundário (só as colunas de ColunasATrazer passam).
      inputs: [ext_itens, ext_pedidos]
      options:
        ChavesPrimario: [PEDIDO_ID]
        ChavesSecundario: [PEDIDO_ID]
        ColunasATrazer: [CLIENTE_ID, DATA_EMISSAO, VALOR_TOTAL]
        TipoJoin: LEFT
      x: 350
      y: 300

    - internalId: load_dw
      kind: InsertDatawarehouse
      description: Grava no Data Warehouse
      inputs: [join_itens]
      options:
        # Preenchido por: lumo new table --from-flow flow:44531 --node join_itens
        TableID: 44940
        Mode: Datawarehouse
        PartitionType: NONE
        PrimaryKeys: []
      x: 600
      y: 300

  Connections:
    - from: ext_itens
      to: join_itens
    - from: ext_pedidos
      to: join_itens
    - from: join_itens
      to: load_dw

table:
  id: 44940
  nome: fato_vendas
tags: [vendas]

Os nós de extração convertem os nomes das colunas para MAIÚSCULAS. Todo nó a jusante (Join, SQLProcessor, InsertDatawarehouse) precisa se referir a elas em maiúsculas.

Especificação: header

`id`

Tipo: integer (>= 1) · Obrigatório: sim

Id do flow no servidor. O lumo new flow cria o recurso e escreve o id aqui. Não edite.

`kind`

Tipo: string · Obrigatório: sim · Valor: flow

Discrimina o tipo de recurso e decide contra qual schema o lumo lint valida o arquivo.

`lumo`

Tipo: string · Obrigatório: sim · Valor: v2

Versão do formato de workspace.

`tenantId`

Tipo: integer (>= 1) · Obrigatório: sim

Tenant dono do flow. Vem do lumo init <tenant-id>.

Especificação: body

`nome`

Tipo: string (mínimo 1 caractere) · Obrigatório: sim

Nome exibido do flow.

yaml

nome: Fato Vendas

`load_type`

Tipo: string · Obrigatório: sim · Valores: Total, Incremental, Temporal

Estratégia de carga.

Valor	Comportamento
`Total`	Apaga tudo e recarrega tudo. Use em dimensões e tabelas pequenas.
`Incremental`	Só traz o que é novo desde a última execução. Injeta `{LastDataPoint}` no SQL.
`Temporal`	Recarrega uma janela de datas. Injeta `{StartDate}` e `{EndDate}` no SQL.

`load_type_column`

Tipo: string ou null · Obrigatório: quando load_type é Incremental ou Temporal · Default: null

Coluna de data que delimita a janela de carga. Escolha uma coluna imutável, como a data de emissão. Uma coluna que a origem reescreve, como data de atualização, faz a carga perder registros.

yaml

load_type: Temporal
load_type_column: DATA_EMISSAO

`key_sweep`

Tipo: objeto ou null · Obrigatório: não · Default: null

Consulta da sincronização de exclusões: um SELECT que devolve apenas as chaves ainda existentes na origem. Um agendamento com context: KeySweep roda essa consulta e marca como excluído, na tabela de destino, tudo que não voltou.

null ou ausente significa que não há varredura configurada.

Campo	Tipo	Obrigatório	O que é
`sql`	string (mínimo 1 caractere)	sim	`SELECT` devolvendo só as colunas de chave do destino.
`credentialId`	integer (>= 1)	sim	Credencial que executa a consulta, à parte da que o flow usa para extrair.

yaml

load_type: Incremental
load_type_column: ATUALIZADO_EM

key_sweep:
  # Sem WHERE, sem LIMIT: o conjunto vivo inteiro.
  sql: SELECT id AS PEDIDO_ID FROM pedidos
  credentialId: 317

Só é válido quando load_type é Incremental e a tabela de destino é table_type: table com key_type: unique e key_columns preenchido. A marcação de exclusão substitui a linha pela chave; sem chave única ela acrescentaria uma linha morta ao lado da viva, em vez de substituí-la.

O lumo lint confere o formato. Quem confere se o SQL é somente leitura é o servidor, no push.

WARNING

A consulta precisa devolver todas as chaves que uma carga Total traria, sem filtro de período. Tudo que não vier é marcado como excluído, sem erro nenhum, e o histórico perdido só volta com uma carga Total.

Os nomes das colunas precisam bater com as colunas de chave do destino (use alias quando diferirem). Formato de chave que não casa é silenciosamente catastrófico, então teste a consulta no editor web antes de agendar.

Exige agente 1.27.0 ou mais novo.

`tokenId`

Tipo: integer ou null · Obrigatório: não · Default: null

Agente que executa o flow. null usa o primeiro agente ativo do tenant. Em tenant com vários agentes, aponte o agente explicitamente. Liste os disponíveis com lumo list agent.

Para migrar um flow de agente, edite este campo e dê lumo push.

`development_variables`

Tipo: array · Obrigatório: não · Default: []

Variáveis de entrada passadas aos nós em cada execução.

`nodes`

Tipo: object · Obrigatório: sim

Contém o DAG. Exige as duas chaves, mesmo vazias.

yaml

nodes:
  Processors: []
  Connections: []

`Processors`

Tipo: array de objetos · Obrigatório: sim

Os nós do flow. Cada item exige internalId e kind. Veja Especificação: processor.

`Connections`

Tipo: array de objetos · Obrigatório: sim

As arestas do DAG. Cada item exige from e to.

`table`

Tipo: object ou null · Obrigatório: não

Tabela do DW que este flow alimenta. O lumo new table --from-flow preenche.

yaml

table:
  id: 44940
  nome: fato_vendas

`owners`

Tipo: array de objetos · Obrigatório: não

Donos do flow, cada item com userId.

`tags`

Tipo: array de string · Obrigatório: não · Default: []

Rótulos livres para organizar o workspace.

Especificação: processor

Cada item de nodes.Processors.

`internalId`

Tipo: string (mínimo 1 caractere) · Obrigatório: sim

Identificador do nó dentro do flow. É o que inputs e Connections usam para se referir a ele. Precisa ser único no flow.

`kind`

Tipo: string (mínimo 1 caractere) · Obrigatório: sim

Tipo do processador, que decide quais chaves options aceita. Os tipos em uso hoje:

Grupo	Valores
Extração	`ExtractPostgreSQL`, `ExtractMySQL`, `ExtractSQLServer`, `ExtractOracleDB`, `ExtractFirebird`, `ExtractInformix`, `ExtractInterSystemsIRIS`, `ExtractODBC`, `ExtractBigQuery`, `ExtractDatalake`, `ExtractLakehouse`, `ExtractStaticCSV`, `HTTPRequest`, `AIExtract`
Transformação	`Join`, `Union`, `SQLProcessor`, `PythonProcessor`, `PythonConfigurator`
Carga	`InsertDatawarehouse`

Rode lumo scaffold node <kind> para ver o fragmento YAML de um tipo, com as chaves de options comentadas.

`description`

Tipo: string ou null · Obrigatório: não

Descreve o que o nó faz. Aparece no editor visual.

`inputs`

Tipo: array de string · Obrigatório: não · Default: []

Nós que alimentam este. Cada item é o internalId ou o complete_id de outro nó do mesmo flow. O lumo lint falha se a referência não existir.

Nós de extração têm inputs: []. A ordem importa em Join (o primeiro é o primário) e em Union.

yaml

inputs: [ext_itens, ext_pedidos]

`options`

Tipo: object · Obrigatório: não

Parâmetros do nó. As chaves aceitas dependem de kind, e o schema não as restringe. Consulte lumo scaffold node <kind> ou a referência de processadores.

`metadata`

Tipo: object · Obrigatório: não

Dados livres associados ao nó.

`complete_id`

Tipo: string · Obrigatório: não

Identificador qualificado que o servidor atribui, no formato <Kind>|<internalId>. Aparece depois de um lumo pull. Escrever internalId em inputs é suficiente.

`x`

Tipo: number · Obrigatório: não

Posição horizontal do nó no editor visual.

`y`

Tipo: number · Obrigatório: não

Posição vertical do nó no editor visual.

Especificação: connection

Cada item de nodes.Connections.

`from`

Tipo: string · Obrigatório: sim

internalId do nó de origem.

`to`

Tipo: string · Obrigatório: sim

internalId do nó de destino.

Campos gerenciados pelo servidor

O servidor calcula estes campos e ignora edições no push. Eles aparecem depois de um lumo pull.

Campo	O que é
`_state`	`draft`, `published` ou `inconsistent`.
`deskId`	Desk em que o flow foi publicado. Mude com `lumo flow publish`.
`version`	Versão do recurso.
`cloned_from`	Flow de origem, quando o flow veio de um `lumo clone`.
`originalTableId`	Tabela de origem do clone.
`criado_em`, `criado_por`, `publicado_em`, `publicado_por`	Auditoria.

Flow ​

Modelo de configuração ​

Configuração completa ​

Especificação: header ​

id ​

kind ​

lumo ​

tenantId ​

Especificação: body ​

nome ​

load_type ​

load_type_column ​

key_sweep ​

tokenId ​

development_variables ​

nodes ​

Processors ​

Connections ​

table ​

owners ​

tags ​

Especificação: processor ​

internalId ​

kind ​

description ​

inputs ​

options ​

metadata ​

complete_id ​

x ​

y ​

Especificação: connection ​

from ​

to ​

Campos gerenciados pelo servidor ​

Flow

Modelo de configuração

Configuração completa

Especificação: header

`id`

`kind`

`lumo`

`tenantId`

Especificação: body

`nome`

`load_type`

`load_type_column`

`key_sweep`

`tokenId`

`development_variables`

`nodes`

`Processors`

`Connections`

`table`

`owners`

`tags`

Especificação: processor

`internalId`

`kind`

`description`

`inputs`

`options`

`metadata`

`complete_id`

`x`

`y`

Especificação: connection

`from`

`to`

Campos gerenciados pelo servidor