flow

Tipo

Referência

Para

Desenvolvedor, Engenharia de dados

Um flow é um Dataflow: extrai de uma origem, transforma e grava numa tabela do HorusDW. O arquivo fica em flows/<slug>--<id>.yaml. Os dois eixos da carga, Extração e Gravação, estão explicados em tipos de carga.

Crie com lumo new flow --name "Fato Vendas", edite o arquivo que o comando escreveu e mande com lumo push flow:<id>. O push grava o rascunho; para publicar numa Mesa de Dados, lumo flow publish flow:<id> --desk <id>.

Campos

• fileName
  string · opcional · valores: regex ^[a-z0-9-]+--\d+\.yaml$
• header
  object · obrigatório
• header.id
  integer · obrigatório · valores: ≥ 1
• header.kind
  string · obrigatório · valores: flow (fixo)
• header.lumo
  string · obrigatório · valores: v2 (fixo)
• header.tenantId
  integer · obrigatório · valores: ≥ 1
• header.appId
  integer · opcional · valores: ≥ 1
• header.dashboardId
  integer · opcional · valores: ≥ 1
• body
  object · obrigatório
• body.criado_em
  string ou null · opcional · valores: regex ^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?Z$
  Somente leitura: o servidor preenche.
• body.criado_por
  string ou integer ou null · opcional
  Somente leitura: o servidor preenche.
• body.publicado_em
  string ou null · opcional · valores: regex ^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?Z$
  Somente leitura: o servidor preenche.
• body.publicado_por
  string ou integer ou null · opcional
  Somente leitura: o servidor preenche.
• body.nome
  string · obrigatório · valores: não vazio
• body._state
  string · opcional · valores: draft, published, inconsistent
  Somente leitura: o servidor preenche. Derived lifecycle state. draft=deskId null (work-in-progress, owner-only). published=deskId set (live, readonly, shared). inconsistent=domain invariant violated (e.g. published flow ↔ draft table); treat as 'stop and ask'. Server-computed; edits are ignored on push.
• body.load_type
  string · obrigatório · valores: Total, Incremental, Temporal
• body.load_type_column
  string ou null · opcional
• body.write_mode
  string ou null · opcional · valores: replace_all, replace_window, upsert_by_key, insert_only
  Axis 2 of the load, the write mode: what happens in the destination to what the extraction brought. replace_all deletes everything before inserting; replace_window deletes the window and needs load_type_column; upsert_by_key overwrites the row with the same key and requires key_type unique with key_columns on a table_type=table destination; insert_only only appends, and is refused on a destination that already deduplicates. replace_all and replace_window are the only ones that notice a record deleted at the source. null or absent = derived from load_type plus the destination's key_type, which is how every flow written before this field runs. Only five of the twelve combinations with load_type are valid: lumo lint refuses the ones that would corrupt data, and names which of the two fields to change.
• body.key_sweep
  object ou null · opcional
  Deletion sync ("varredura de exclusões"): a query returning ONLY the keys still present in the source. A schedule entry with context=KeySweep runs it and marks as deleted, in the destination table, every row whose key did not come back. null (or absent) means no sweep is configured. Only valid on load_type=Incremental whose destination is a table_type=table with key_type=unique and key_columns set - the delete mark is a Doris unique-model concept, and without a key it would add a dead row beside the live one instead of replacing it. The query is refused by the server unless it is read-only; lumo lint checks the shape, the server owns the SQL safety check.
• body.key_sweep.sql
  string · obrigatório · valores: não vazio
  SELECT returning only the key columns, in the same order and type as the destination's key_columns. A mismatched key format is silently catastrophic (VARCHAR vs INT, trailing spaces, different case), so test it in the web editor before scheduling.
• body.key_sweep.credentialId
  integer · obrigatório · valores: ≥ 1
  Credential used to run the key query. Its own field rather than the flow extractor's: the query needs nothing beyond read access, and a read-only credential is the real defense here.
• body.deskId
  integer ou null · opcional · valores: ≥ 1
  Somente leitura: o servidor preenche. Server-managed publish state; change through publish commands, not declarative push.
• body.cloned_from
  integer ou null · opcional · valores: ≥ 1
  Somente leitura: o servidor preenche. Server-managed clone lineage.
• body.originalTableId
  integer ou null · opcional · valores: ≥ 1
  Somente leitura: o servidor preenche. Server-managed clone source table.
• body.version
  integer ou null · opcional · valores: ≥ 0
  Somente leitura: o servidor preenche. Server-managed resource version.
• body.tokenId
  integer ou null · opcional · valores: ≥ 1
  Editable canonical agent/token assignment for this flow.
• body.development_variables
  array · opcional
• body.nodes
  object · obrigatório
• body.nodes.Connections
  array de object · obrigatório
• body.nodes.Connections[].from
  string · obrigatório
• body.nodes.Connections[].to
  string · obrigatório
• body.nodes.Processors
  array de object · obrigatório
• body.nodes.Processors[].complete_id
  string · opcional
• body.nodes.Processors[].internalId
  string · obrigatório · valores: não vazio
• body.nodes.Processors[].kind
  string · obrigatório · valores: não vazio
• body.nodes.Processors[].description
  string ou null · opcional
• body.nodes.Processors[].inputs
  array de string · opcional
• body.nodes.Processors[].metadata
  object · opcional
• body.nodes.Processors[].options
  object · opcional
• body.nodes.Processors[].x
  number · opcional
• body.nodes.Processors[].y
  number · opcional
• body.table
  object ou null · opcional
• body.table.id
  integer · opcional · valores: ≥ 1
• body.table.nome
  string · opcional
• body.tags
  array de string · opcional
• body.owners
  array de object · opcional
• body.owners[].userId
  integer · opcional · valores: ≥ 1

Exemplo

Um fato de vendas que recarrega uma janela de datas a cada execução.

# flows/fato-vendas--5101.yaml
# yaml-language-server: $schema=https://docs.horusbi.com.br/schemas/v2/flow.schema.json
id: 5101
kind: flow
lumo: v2
tenantId: 1200
---
nome: Fato Vendas
# Extração: uma janela de datas. A coluna precisa ser imutável na origem.
load_type: Temporal
load_type_column: DATA_EMISSAO
# Gravação: apaga a janela no destino e insere de novo.
write_mode: replace_window
tokenId: 41                      # Agente ETL que roda o Dataflow (lumo list agent)
nodes:
  Processors:
    - internalId: ext_pedidos
      kind: ExtractPostgreSQL
      inputs: []
      options:
        Credential: erp-producao # chave da credencial, não o id (lumo list credential)
        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}'
    - internalId: load_dw
      kind: InsertDatawarehouse
      inputs: [ext_pedidos]
      options:
        TableID: 6230            # preenchido por lumo new table --from-flow
        Mode: Datawarehouse
  Connections:
    - from: ext_pedidos
      to: load_dw

lumo scaffold flow lista os modelos prontos (PostgreSQL, CSV, HTTP, Python, Join, Union e outros), e lumo scaffold node <kind> mostra as options de cada tipo de nó, comentadas.

Combinações de Extração e Gravação

Das 12 combinações de load_type com write_mode, 5 são aceitas. O lumo lint recusa as outras e diz qual dos dois campos trocar.

load_type	write_mode aceitos
Total	replace_all
Temporal	replace_window, upsert_by_key
Incremental	upsert_by_key, insert_only

write_mode ausente faz o servidor deduzir a Gravação de load_type e do key_type da tabela de destino, como rodavam os Dataflows anteriores ao campo. upsert_by_key exige destino table_type: table com key_type: unique e key_columns, e insert_only é recusado num destino que já tira duplicata. Só replace_all e replace_window percebem um registro apagado na origem; para os outros, veja exclusões na origem e o campo key_sweep.

Limites

• Os nós de extração devolvem os nomes de coluna em maiúsculas. Os nós seguintes precisam usar maiúsculas.
• Uma aresta aparece em dois lugares: em inputs do nó de destino e em Connections. Os dois precisam concordar, e o lint recusa um inputs que aponta para nó inexistente.
• No Join, a ordem de inputs importa: o primeiro é o principal, e todas as colunas dele ficam. O Union casa as colunas pelo nome.
• Credential leva a chave da credencial (o campo key do arquivo dela), não o id nem o nome.
• Não edite TableID à mão. O lumo new table --from-flow e o lumo flow sync-table o preenchem; um valor escrito à mão gera drift.
• Mode do InsertDatawarehouse precisa combinar com o table_type da tabela (Datawarehouse para table, Datalake para cloud). Discordando, a carga grava no lugar certo, mas a carga Incremental procura a marca da última execução no lugar errado, sem erro.
• No HorusDW (table_type: table), a opção PrimaryKeys do InsertDatawarehouse é ignorada: quem tira duplicata é o key_type: unique da tabela.
• key_sweep só vale com load_type: Incremental e destino com key_type: unique. A consulta precisa devolver todas as chaves vivas na origem, sem filtro de período: o que não voltar é marcado como apagado.
• Trocar tokenId e dar push muda o Agente ETL do Dataflow e leva os agendamentos dele junto. Se a credencial do Dataflow não está no agente destino, o push é recusado e nada é gravado: rode lumo credential transfer antes.
• Dataflow publicado não aceita push: lumo flow clone, edite o clone e publique.
• lumo flow run roda só Dataflow publicado. Num rascunho, teste nó a nó com lumo flow run-node.