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
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

`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 ​

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`

`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