Configurador Python

O Configurador Python calcula variáveis. Ele roda um script, e tudo que estiver no dicionário variables no fim vira variável para os nós seguintes. O dado do fluxo passa por outros nós; este só produz valores.

import datetime

ontem = datetime.date.today() - datetime.timedelta(days=1)

variables = {
    "DataCorte": ontem.strftime("%Y-%m-%d"),
    "Filiais": "1,2,7",
}

Depois disso, {DataCorte} e {Filiais} funcionam em qualquer nó posterior, e em Python chegam como DataCorte e Filiais. Ver Variáveis para as três formas de consumir uma.

Serve para o que a configuração pronta não cobre. Regra de data que depende de feriado, lista de filiais que vem de uma consulta, token buscado antes da carga, endereço de banco conhecido só na hora: tudo isso vira variável, e o resto do Dataflow consome como se sempre tivesse existido.

Ele roda antes do resto

O Agente ETL trata o Configurador como caso à parte. Havendo um no fluxo, ele e tudo de que ele depende executam antes do resto, num escalonamento próprio.

Duas consequências práticas.

Configurador sem entrada não precisa de ligação. Você não precisa ligá-lo aos extratores para garantir que rode primeiro. Ele já roda.

Configurador pode ter entradas. Ligue um ou mais nós a ele e cada resultado chega como um DataFrame numerado a partir de zero: input_0, input_1, e assim por diante, na ordem das ligações. Essas consultas também são puxadas para a frente do fluxo, junto com ele. É o que permite ler uma tabela de configuração no banco e transformar o resultado numa variável antes de qualquer carga começar.

import pandas as pd

# input_0 é o resultado do primeiro nó ligado à entrada
codigos = ",".join(input_0["CODIGO"].astype(str))

variables = {"CODIGOS": codigos}

Buffer de reprocessamento no ponto de corte

Numa Extração Incremental o Agente injeta {LastDataPoint}, o maior valor que a coluna de controle já atingiu no destino. Cortar exatamente nele é o caso simples, e funciona enquanto a origem grava em ordem.

Nem toda origem grava em ordem. Transação longa, fuso configurado errado, integração que reprocessa em lote: em todos, um registro pode nascer com valor de controle abaixo de um que já foi carregado. Cortando no ponto exato, esse registro nunca entra, e ninguém percebe, porque a carga termina com sucesso.

A saída é recuar o corte de propósito. Um Configurador calcula a data recuada e o extrator usa a variável dele:

import datetime

corte = LastDataPoint - datetime.timedelta(days=3)

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

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

O recuo traz registros que já estavam no destino. Isso só é seguro com a Gravação Atualizar por chave, que substitui a linha repetida em vez de acrescentá-la. Com Somente inserir, o mesmo recuo duplica tudo que ele alcançar.

O tamanho do recuo depende da origem. Duas perguntas resolvem: quanto tempo a transação mais longa dela fica aberta, e qual o pior desvio de relógio que você já viu ali. Recuo maior custa reprocessamento a cada execução, então vale medir em vez de arredondar para cima.

IMPORTANT

O tipo de {LastDataPoint} vem da coluna de controle no destino: coluna de data chega como data, coluna numérica chega como número, e as demais chegam como texto. Escreva o script para o tipo da sua coluna. Num controle numérico o recuo é uma subtração (LastDataPoint - 1000); num controle de texto, recuar não faz sentido, e o caminho é outro.

Conexão montada em tempo de execução

O caminho normal de conexão é o cadastro em Conexões de Banco. Quando o endereço só é conhecido na hora, um Configurador monta a string e o extrator aponta para a variável.

Na tela do extrator, marque usar variável de conexão e informe o nome da variável. O nó passa a resolver a conexão pelo valor dela na execução.

# Segredos vêm do ambiente da máquina onde o Agente roda, com o prefixo ENV_.
# Nunca escreva senha dentro do script nem numa variável do Dataflow: o texto
# fica salvo na definição do fluxo e acompanha toda cópia dela.
connection_string = (
    f"Host={ENV_DB_HOST};Uid={ENV_DB_USER};Password={ENV_DB_PASS};Database=vendas"
)

variables = {"CS": connection_string}

Ver Variáveis para como o prefixo ENV_ funciona.

Um nó lendo vários bancos

O campo de conexão por variável aceita várias conexões separadas por vírgula. O extrator roda a mesma consulta em cada uma e empilha os resultados. É assim que um fluxo só atende uma rede de lojas em que cada loja tem o próprio banco.

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

def connection_string(db):
    return f"Server={ENV_DB_HOST};Port=3306;Database={db};Uid={ENV_DB_UID};Pwd={ENV_DB_PWD}"

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

WARNING

A vírgula separa as conexões, então vírgula literal dentro de um valor precisa ser escapada com \,. Uma senha com vírgula sem escape quebra a lista em duas conexões inválidas, e o erro que aparece é de conexão, não de sintaxe.

Aqui a lista de bancos veio de DB_LIST, uma variável global do tenant: acrescentar uma loja é editar uma variável, e nenhum fluxo muda.

SQL gerado a partir de um de-para

Quando a mesma informação mora em tabelas de nome diferente em cada empresa, a consulta não pode ser fixa. O padrão é guardar o de-para numa tabela, lê-lo com uma consulta ligada à entrada do Configurador, e gerar a consulta final a partir de um template.

from re import sub

template = """
SELECT '[EMP]' AS EMPRESA, p.ID AS PEDIDO, p.DATA_EMISSAO
FROM {PEDIDO} p
"""

# input_0 traz o de-para: uma linha por empresa e tabela lógica
mapa = {f"{r['EMPRESA']}-{r['LOGICA']}": r['FISICA'] for _, r in input_0.iterrows()}

def para_empresa(empresa):
    texto = template.replace("[EMP]", empresa)
    return sub(r"\{(.*?)\}", lambda m: mapa[f"{empresa}-{m.group(1)}"], texto)

empresas = input_0["EMPRESA"].unique()

variables = {
    "SQLConsulta": " UNION ALL ".join(para_empresa(e) for e in empresas)
}

O extrator recebe {SQLConsulta} como consulta. Acrescentar uma empresa passa a ser inserir linhas no de-para, sem tocar em nenhum fluxo.

Quando não usar

Boa parte do que se resolve com script já tem campo próprio, e o campo é mais fácil de manter.

Para recortar por período, use Extração Janela, que já injeta {StartDate} e {EndDate}. Para não duplicar registro repetido, use Gravação Atualizar por chave. Para um valor que se repete em vários fluxos e muda de uma vez só, use variável global.

Cada script acrescenta uma linha de código que alguém vai ter que entender depois. Vale gastá-la no que só ele resolve.

Falhas comuns

"Nenhuma variável foi configurada pelo script". O script terminou sem definir variables. A execução falha de propósito, em vez de seguir com valor vazio que ninguém veria.

A variável chega literal no banco. Um {CortePonderado} que aparece na mensagem de erro do banco significa que a variável não existia naquela execução. Confira o nome, e confira que o Configurador está no mesmo fluxo.

A carga duplicou depois de acrescentar o recuo. O recuo trouxe registros que já estavam lá e a Gravação não substitui por chave. Ver Buffer de reprocessamento.

Erro de conexão depois de mexer na senha. Senha com vírgula sem escape. Ver Um nó lendo vários bancos.

Relacionados

• Variáveis: as origens, as três formas de escrever, e o que o sistema injeta sozinho
• Tipos de Carga: Extração, Gravação, e o que {LastDataPoint} significa
• Conexões de Banco: o caminho normal, por cadastro