Calcular variáveis na hora da execução

Tipo

Como fazer

Para

Engenharia de dados

Na tela

HorusETL › Data Flows › editor do Dataflow › Configurador

Com um nó Configurador, o Dataflow calcula variáveis antes de o resto rodar, e os nós seguintes as usam como {NOME}. Use quando o valor depende da execução: um ponto de corte recuado, a string de conexão de um banco que só se conhece na hora, o token de uma API que expira. As variáveis que o Lumo já entrega prontas estão em Variáveis.

Antes de começar

Escolha o Configurador:

Configurador	Use para	Agente ETL mínimo
Python	cálculo com bibliotecas, valor lido de uma tabela, várias chamadas	qualquer versão
JavaScript	montar texto (conexão, SQL, datas) dentro do próprio Agente, sem rede	1.29.14
HTTP	uma chamada de API cujo resultado vira variável por JSONPath	1.29.0

Para recortar por período ou pegar o que mudou, a Extração Janela ou Incremental já entrega {StartDate}, {EndDate} e {LastDataPoint}; não precisa de Configurador.

Passos

1. No editor do Dataflow, abra o grupo Configurador da paleta e arraste o Configurador escolhido.
2. Se o cálculo depende de uma tabela, ligue o nó que lê essa tabela à entrada do Configurador (Python e JavaScript). Cada entrada chega como input_0, input_1, na ordem das ligações, com as colunas em maiúsculas.
3. Escreva o script, ou preencha a chamada no Configurador HTTP.
4. Use as variáveis nos outros nós, com chaves: {CortePonderado}. Não é preciso ligar o Configurador aos nós que usam as variáveis; ele roda antes de todos.
5. Clique em Executar Todos os Nós para testar. As variáveis devolvidas ficam nas Variáveis de Desenvolvimento do Dataflow, para os nós seguintes poderem ser testados.

Recuar o ponto de corte

Uma origem que grava fora de ordem (transação longa, integração em lote) pode criar um registro com atualizado_em menor que o ponto de corte de uma carga Incremental. Cortando exatamente em {LastDataPoint}, esse registro nunca entra. Recue o corte alguns dias:

Python

import datetime

corte = LastDataPoint - datetime.timedelta(days=3)
variables = {"CortePonderado": corte.strftime("%Y-%m-%d %H:%M:%S")}

JavaScript

const corte = Temporal.PlainDateTime.from(LastDataPoint).subtract({ days: 3 });
variables = {
  CortePonderado: corte.toString({ smallestUnit: "second" }).replace("T", " "),
};

SELECT * FROM pedidos WHERE atualizado_em > '{CortePonderado}'

O recuo traz de novo linhas que já estão no destino, então só use com a Gravação Atualizar por chave. No Python, LastDataPoint chega como datetime quando a coluna de corte é data; no JavaScript, chega como texto. Numa coluna de corte numérica, o recuo é uma subtração: LastDataPoint - 1000.

Montar a conexão na hora

Quando o endereço do banco só se conhece na execução, o Configurador monta a string e o nó de banco a usa com Conexão por variável, informando o nome da variável em Variável a usar. Senha vem de variável protegida ou do ambiente do Agente (ENV_), nunca escrita no script.

Uma lista separada por vírgula faz o nó rodar a mesma consulta em cada banco e juntar os resultados. Vírgula dentro de um valor precisa de escape, \,:

Python

bancos = [b.strip() for b in DB_LIST.split(",") if b.strip()]

def conexao(db):
    return f"Server={ENV_DB_HOST};Database={db};Uid={ENV_DB_USER};Pwd={ENV_DB_PASS}"

variables = {"LOJAS_CS": ",".join(conexao(b).replace(",", "\\,") for b in bancos)}

JavaScript

const bancos = DB_LIST.split(",").map(b => b.trim()).filter(b => b);
const conexao = db =>
  `Server=${ENV_DB_HOST};Database=${db};Uid=${ENV_DB_USER};Pwd=${ENV_DB_PASS}`;

variables = { LOJAS_CS: bancos.map(b => conexao(b).replaceAll(",", "\\,")).join(",") };

DB_LIST é uma variável global com a lista de bancos: acrescentar uma loja vira editar a variável, sem mexer no Dataflow.

Buscar um token de API

Para uma API que pede login antes de cada consulta, o Configurador HTTP faz a chamada e guarda o campo da resposta. Um login OAuth2 client_credentials:

1. Na aba Requisição, método POST, a URL do endpoint de token e, no corpo, grant_type, client_id e client_secret={ENV_PARCEIRO_SECRET}.
2. Na aba Variáveis, clique em Adicionar variável e preencha ACCESSTOKEN com o JSONPath $.access_token.
3. No nó Requisição HTTP que busca os dados, na aba Autenticação, escolha Bearer Token e escreva {ACCESSTOKEN} no token.

Cada execução pede um token novo. Se o provedor invalida o token anterior ao emitir outro, dois Dataflows rodando ao mesmo tempo derrubam um ao outro; espace os agendamentos.

Ligue a aba Retry só depois de ver o login funcionar: ela repete qualquer falha, inclusive 401 e 403, e com o segredo errado e 3 tentativas o endpoint de login recebe 4 chamadas inválidas por execução, o que alguns provedores punem bloqueando a credencial. Testar o nó no editor grava o token nas Variáveis de Desenvolvimento do Dataflow. O Configurador HTTP não guarda nada entre execuções, então um refresh token que muda a cada renovação pede o Configurador Python, com flow_state.

Parar o Dataflow sem erro

Um Configurador que devolve SkipFlow com valor "true" encerra o Dataflow com sucesso, sem rodar os outros nós. Serve, por exemplo, para não carregar nada quando a origem ainda não fechou o dia.

Se der errado

Nenhuma variável foi configurada pelo script. O script terminou sem preencher variables. A execução falha de propósito, em vez de seguir com valor vazio.

A variável aparece literal no banco, como {CortePonderado}. Ela não existia naquela execução. Confira o nome, com maiúsculas, e se o Configurador está no mesmo Dataflow.

Erro de conexão depois de trocar a senha. A senha tem vírgula sem escape e virou duas conexões.

Relacionado

• Configuradores: entradas, saídas, limites e mensagens de cada um.
• Variáveis
• Requisição HTTP

Pelo Lumo CLI

Este recurso também pode ser criado e versionado em YAML. Veja a referência de variables.