Cargas incrementais e upserts

Uma recarga completa é simples e, a partir de certo tamanho, impossível. Estes são os padrões que a substituem.

Merge em vez de overwrite

{
  "output": {
    "format": "delta",
    "path": "analytics.orders",
    "mode": "merge",
    "options": {
      "merge_keys": ["order_id"],
      "merge_condition": "T.deleted = false"
    }
  }
}

T é a tabela de destino, S o DataFrame de entrada. merge_condition adiciona SQL à cláusula de correspondência — útil para ignorar linhas com soft-delete ou restringir o merge a uma partição.

Uma linha por chave

Um merge falha quando o DataFrame de entrada tem mais de uma linha por chave — o Delta lança multiple source rows matched. Deduplique antes de gravar e adicione uma validação unique nas chaves de merge para que o pipeline reporte o problema em vez da camada de armazenamento.

Leia apenas a nova janela

Empurre a janela para dentro da origem para que a leitura seja pequena:

// a partitioned table or a database query
{ "type": "filter", "condition": "ordered_at >= '{since}'" }

fw.run("orders.json", params={"since": last_success_timestamp})

Manter since fora do arquivo — no orquestrador, em uma tabela de controle — é o que torna o pipeline reutilizável tanto para o delta noturno quanto para um backfill.

Encerre quando não há nada a fazer

[
  { "type": "filter", "condition": "updated_at > '{since}'" },
  { "type": "stop_if_empty", "message": "No changes since {since}" },
  { "type": "checkpoint" }
]

As transformações restantes e toda escrita são puladas, e a execução retorna com skipped: true e success: true. Colocado logo após o filtro de janela — antes de joins e da construção de payload — ele transforma uma noite ociosa em alguns segundos em vez de um pipeline completo sobre zero linhas.

result = fw.run("orders.json", params={"since": since})
if result.skipped:
    log.info("nothing to load")     # not a failure

Torne as reexecuções seguras

Um job incremental será reexecutado — após uma falha, após um arquivo que chegou atrasado, por alguém depurando. Duas propriedades tornam isso inofensivo:

• Escritas idempotentes. merge sobre uma chave de negócio estável produz a mesma tabela quer rode uma vez ou cinco. append não.
• Uma janela determinística. Derive-a de um parâmetro ou de uma tabela de controle, nunca de now() dentro do pipeline, para que uma reexecução cubra o mesmo intervalo.

{ "type": "drop_duplicates", "columns": ["order_id"] }

Deduplicar pela chave de merge logo antes da escrita fecha a última brecha: uma origem que entrega o mesmo registro duas vezes dentro de uma janela.

Staging, depois commit

Quando vários pipelines contribuem para um destino, deixe cada um gravar uma view de staging e dê ao pipeline final a tarefa de validar e publicar:

// pipelines 1..N
{ "output": { "format": "view", "path": "orders_staging", "mode": "overwrite" } }

// the commit pipeline
{
  "input": { "format": "view", "path": "orders_staging" },
  "validations": {
    "on_failure": "fail",
    "rules": [
      { "type": "unique", "columns": ["order_id"] },
      { "type": "row_count", "min": 1 }
    ]
  },
  "outputs": [
    { "format": "delta", "path": "analytics.orders", "mode": "merge",
      "options": { "merge_keys": ["order_id"] } }
  ]
}

fw = Sparquet()
for conf in ["conf_a.json", "conf_b.json", "conf_c.json"]:
    result = fw.run(conf, params={"since": since})
    if not result.success:
        raise RuntimeError(f"{conf}: {result.error}")

commit = fw.run("conf_commit.json")
fw.stop()

Nada chega ao destino até que todo o conjunto tenha sucesso e as regras passem. As temp views vivem na sessão Spark, então todo pipeline deve rodar no mesmo processo — que é exatamente o que reutilizar um único Sparquet oferece.

Fique de olho na linhagem

Um pipeline incremental que faz join de várias origens constrói um plano profundo. Duas regras o mantêm barato:

• checkpoint após os joins pesados, antes do collect e antes de se ramificar para os destinos.
• Empurre a lista de chaves para dentro de leituras grandes com pushdown em runtime para que a origem pule arquivos em vez de escaneá-los.

Escolhendo um padrão

Situação	Padrão
Tabela de dimensão pequena	overwrite completo — a simplicidade vence
Tabela fato com chave estável	merge na chave
Log de eventos apenas-append	append mais uma leitura que deduplica
Vários jobs alimentando uma tabela	view de staging, depois um pipeline de commit
Backfill	o mesmo arquivo, um {since} diferente