O que significa idempotente em pipelines de dados? Como evitar execuções de ETL duplicadas com Apache Beam

Idempotente significa que você pode rodar uma operação uma ou muitas vezes e obter o mesmo resultado. Em um pipeline de dados, isso significa que um retry, uma nova execução ou um backfill deixa a tabela de destino exatamente como uma execução limpa deixaria.

Se um retry acabou de duplicar as suas linhas, o seu pipeline não era idempotente quando as gravou. O Apache Beam refaz o trabalho automaticamente e garante resultados exactly-once dentro do pipeline. Mas continua sendo responsabilidade sua garantir que aquilo que o seu código grava em um banco de dados ou arquivo seja seguro.

O que significa idempotente?

Idempotente significa que fazer a mesma operação muitas vezes tem o mesmo efeito que fazê-la uma vez só. Uma execução de pipeline é idempotente quando rodá-la duas vezes para a mesma janela de entrada deixa a tabela de destino exatamente no mesmo estado que rodá-la uma vez.

O HTTP tem o exemplo cotidiano mais claro. Um método é idempotente quando muitas requisições idênticas têm o mesmo efeito pretendido no servidor que uma única requisição. PUT, DELETE e GET são idempotentes. POST não é.

A diferença está no que cada requisição envia. O corpo de um PUT é o estado final em que você quer que o recurso fique, então enviá-lo de novo não muda nada. O corpo de um POST é uma informação para o servidor processar, então enviá-lo de novo pode criar um segundo registro. Os clientes podem repetir requisições idempotentes automaticamente depois de uma falha de rede, já que repeti-las não causa problema. Cargas de dados podem se comportar como qualquer um dos dois métodos:

• Sobrescrever a partição de 2026-10-12 funciona como um PUT. A segunda execução substitui as linhas pelas mesmas linhas.
• Um INSERT simples das linhas daquele dia funciona como um POST. A segunda execução adiciona uma segunda cópia.

O Apache Beam espera o mesmo do seu código. O guia de programação dele recomenda que cada função possa ser repetida ou refeita quantas vezes for necessário sem causar efeitos colaterais indesejados, porque o Beam não promete quantas vezes o seu código vai rodar.

Por que pipelines de ETL gravam linhas duplicadas?

Pipelines de ETL gravam linhas duplicadas quando algum trabalho roda mais de uma vez e a gravação final adiciona em vez de substituir. Retries e novas execuções são normais, então as duplicatas vêm da gravação, não do retry.

• Retries de bundles dentro do runner. O Beam divide uma coleção em bundles. Se um elemento de um bundle falha, o bundle inteiro falha e precisa ser refeito, muitas vezes em outro worker. Todos os elementos daquele bundle passam de novo pelo seu código, inclusive os que já tinham dado certo.
• Entrega at-least-once. Jobs de streaming no Dataflow podem rodar em um modo at-least-once que custa menos e tem menor latência, mas permite duplicatas.
• Retries do orquestrador. O Airflow refaz tasks que falharam, e é por isso que o guia dele diz que as tasks devem produzir o mesmo resultado em toda nova execução e devem usar UPSERT em vez de INSERT.
• Backfills e novas execuções manuais. Alguém corrige um bug e roda de novo a semana passada. Cada dia que já tinha sido carregado é carregado outra vez.

O runner limpa a bagunça dos próprios retries. Quando um bundle falha, o Beam descarta a saída dele e a recalcula. Mas essa limpeza só cobre os dados dentro do pipeline. Se o seu código já enviou uma linha para um banco de dados, chamou uma API ou gravou um arquivo, essa gravação fica onde caiu, e o retry pode repeti-la.

[IMAGEM 1: fazer upload de gtm-1361-img1.png aqui]

Código não determinístico piora a situação. Se uma task lê "os dados mais recentes" ou chama now(), uma nova execução processa linhas diferentes das da primeira. Se você amarra cada execução a uma partição fixa, como o data_interval_start do Airflow, toda nova execução recebe a mesma entrada.

O que o Apache Beam garante, e onde essa garantia termina?

O Beam no Dataflow garante resultados exactly-once dentro do pipeline: os registros são processados pelo menos uma vez, e os resultados são confirmados exatamente uma vez. Gravações em um armazenamento externo e outros efeitos colaterais só são exactly-once se o sink tiver suporte a isso.

No Dataflow, pipelines batch sempre usam processamento exactly-once, e pipelines de streaming usam por padrão. Mas, nos dois casos, um mesmo registro ainda pode passar por uma transformação mais de uma vez, ou até rodar em dois workers ao mesmo tempo. Uma DoFn que chama uma API remota pode fazer essa chamada duas vezes para o mesmo registro, então tornar a gravação segura é responsabilidade sua.

Use um sink nativo com gravações exactly-once

No BigQuery, o método de gravação que você escolhe define o que um retry faz:

Veja como a página do BigQueryIO lista o método exactly-once:

[IMAGEM 2: fazer upload de gtm-1361-img2.png aqui]

O WRITE_TRUNCATE substitui a tabela inteira. Sobrescrever uma única partição de data acontece do lado do warehouse, o que vamos ver na próxima seção.

O BigQuery não garante a unicidade da chave primária, então a opção de CDC só funciona quando a chave já é única nos seus dados.

Congele a entrada antes de um efeito colateral

A idempotência quebra quando um retry envia um conjunto diferente de registros. Por exemplo, uma soma por janela pode disparar e depois disparar de novo com um elemento atrasado incluído. Se você grava essa soma duas vezes pela chave, a linha continua estável, mas um sistema downstream que já agiu com base no primeiro valor agiu com o número errado.

Adicionar um Reshuffle antes do efeito colateral resolve isso. Ele ajuda a garantir que só uma versão da saída de uma DoFn passe por uma fronteira de shuffle. Em Java, você também pode marcar a DoFn que grava com @RequiresStableInput, para que um retry veja a mesma entrada que a primeira tentativa.

Em um job de streaming, você coloca o Reshuffle logo antes da etapa que grava no sistema externo:

[IMAGEM 3: fazer upload de gtm-1361-img3.png aqui]

Descarte repetições com o Deduplicate

O Beam tem uma transformação Deduplicate em Java e Python, mas ela é de melhor esforço e limitada por tempo. Em Python, você passa uma duração em processing time ou event time para que o estado não cresça para sempre. Em Java, o padrão é 10 minutos de processing time. Valores em janelas diferentes nunca contam como duplicatas.

Então o Deduplicate é uma ferramenta de streaming para mensagens entregues de novo. Ele não faz nada por um backfill que roda de novo a terça-feira passada. Para novas execuções em batch, a gravação precisa ser idempotente.

Quais padrões tornam uma carga idempotente?

Você pode tornar uma carga idempotente sobrescrevendo a partição, fazendo merge por uma chave, criando IDs determinísticos, deduplicando em uma tabela de staging ou verificando os dados antes de publicá-los. Cada um deles transforma um append em algo mais próximo de um PUT.

[IMAGEM 4: fazer upload de gtm-1361-img4.png aqui]

A sobrescrita de partição é a opção mais simples para jobs batch diários. Uma execução para 2026-10-12 é dona dessa partição. Se você a rodar de novo, ela substitui a partição pelas mesmas linhas, e um backfill da semana passada substitui sete partições. Isso só funciona quando a execução lê uma janela fixa. É por isso que o Airflow recomenda ler e gravar uma partição específica e nunca "os dados mais recentes disponíveis".

O merge é a melhor opção quando chegam atualizações de linhas antigas, como um pedido cujo status muda dias depois. Uma sobrescrita de partição reconstrói todos os dias em que esse pedido aparece, enquanto um merge por order_id atualiza uma única linha. A chave precisa ser única em cada lote que você faz merge, então deduplique o lote antes se a fonte puder repetir uma linha.

IDs determinísticos são o que faz os outros padrões funcionarem no Beam. Se você gera um hash da chave de negócio (ID do pedido mais data), todo retry recebe o mesmo ID. Uma etapa do Beam pode então agrupar por esse ID e manter uma linha, e o warehouse pode fazer merge por ele.

Os padrões em duas etapas, staging e write-audit-publish, adicionam uma etapa extra, mas são mais seguros. Eles fazem sentido quando uma carga ruim custa mais caro do que uma carga lenta.

Exemplo: rodando um pipeline Beam duas vezes

Se você rodar o mesmo job do Beam duas vezes, dá para ver a diferença em poucas linhas. A versão com append duplica as linhas na segunda execução, enquanto a versão idempotente grava as mesmas duas linhas nas duas vezes.

A demonstração roda localmente no DirectRunner, então você pode rodá-la na sua própria máquina. Ela usa três pedidos de 2026-10-12, e a fonte envia o pedido 2 duas vezes. Instale o Beam primeiro:

pip install apache-beam

[IMAGEM 5: fazer upload de gtm-1361-img5.png aqui]

O pipeline tem dois caminhos. O run_append adiciona cada linha a um arquivo de dentro de uma DoFn, que é exatamente o efeito colateral contra o qual a documentação do Beam alerta. O run_idempotent gera um hash do ID do pedido e da data como chave, mantém uma linha por chave e grava o arquivo da partição daquela data, substituindo-o a cada execução:

import hashlib
import json
import shutil
from pathlib import Path

import apache_beam as beam

DS = "2026-10-12"
OUT = Path(__file__).parent / "out"
ROWS = [
{"order_id": 1, "ds": DS, "amount": 10},
{"order_id": 2, "ds": DS, "amount": 20},
{"order_id": 2, "ds": DS, "amount": 20},
]


def encode_row(row):
return json.dumps(row, sort_keys=True)


class AppendRow(beam.DoFn):
def process(self, row):
with (OUT / "append.jsonl").open("a", encoding="utf-8") as output:
output.write(encode_row(row) + "\n")
return []


def matches_partition(row, ds):
return row["ds"] == ds


def key_row(row):
identity = f"{row['order_id']}|{row['ds']}"
key = hashlib.sha256(identity.encode("utf-8")).hexdigest()[:16]
return key, row


def pick_row(key_and_rows):
_, rows = key_and_rows
return min(rows, key=encode_row)


def run_append():
with beam.Pipeline(runner="DirectRunner") as pipeline:
pipeline | beam.Create(ROWS) | beam.ParDo(AppendRow())


def run_idempotent(ds):
with beam.Pipeline(runner="DirectRunner") as pipeline:
(
pipeline
| beam.Create(ROWS)
| beam.Filter(matches_partition, ds)
| beam.Map(key_row)
| beam.GroupByKey()
| beam.Map(pick_row)
| beam.Map(encode_row)
| beam.io.WriteToText(
str(OUT / f"ds={ds}" / "part"),
shard_name_template="",
file_name_suffix=".jsonl",
)
)


def count_rows(path):
with path.open(encoding="utf-8") as source:
return sum(1 for _ in source)


def main():
shutil.rmtree(OUT, ignore_errors=True)
OUT.mkdir()
for run in range(1, 3):
run_append()
run_idempotent(DS)
append_count = count_rows(OUT / "append.jsonl")
idempotent_count = count_rows(OUT / f"ds={DS}" / "part.jsonl")
print(
f"run {run}: append rows={append_count} idempotent rows={idempotent_count}"
)


if __name__ == "__main__":
main()

Este é o resultado de rodar os dois jobs duas vezes:

run 1: append rows=3 idempotent rows=2
run 2: append rows=6 idempotent rows=2

O caminho com append erra duas vezes. Ele mantém o pedido duplicado da fonte na primeira execução e depois copia as três linhas de novo na segunda. O caminho idempotente remove a duplicata da fonte usando a chave com hash e substitui a partição na nova execução, então o arquivo tem as mesmas duas linhas depois das duas execuções.

O pick_row ordena as linhas de uma chave e pega a primeira, então a mesma entrada sempre escolhe a mesma linha. E o WriteToText com um template de shard vazio grava um único arquivo e o substitui, o que é uma sobrescrita de partição no nível do arquivo. No BigQuery, as mesmas linhas com chave iriam para um merge ou para uma sobrescrita de partição, em vez de um arquivo.

Como a Erathos lida com novas execuções?

A Erathos, uma plataforma gerenciada de ingestão de dados, lida com novas execuções pelo tipo de sync que você escolhe para cada tabela. O tipo de sync recomendado, Partial Overwrite, funciona como um merge pela chave primária, então uma nova execução desse job atualiza as linhas existentes. Os outros tipos de sync também correspondem aos padrões de antes:

A documentação da Erathos descreve os tipos parciais assim:

[IMAGEM 6: fazer upload de gtm-1361-img6.png aqui]

Os tipos parciais precisam de uma chave primária e de uma coluna date ou datetime para usar como cursor. Para APIs, a Erathos escolhe o cursor por você, e para bancos de dados você o define em cada job.

O Partial Append é a escolha certa quando você quer todas as versões de um registro, como uma trilha de auditoria de mudanças de status. É a escolha errada quando os modelos downstream esperam uma linha por chave.

Cargas baseadas em cursor têm uma lacuna que a idempotência não resolve. Se uma linha é deletada na fonte, ela fica no warehouse para sempre, porque o cursor só encontra linhas novas e atualizadas. Você pode usar a configuração Enable Full On para agendar full refreshes periódicos que limpam essas diferenças. Para PostgreSQL, a Erathos também suporta change data capture (CDC). O CDC lê o write-ahead log, onde o PostgreSQL registra cada insert, update e delete, então os deletes também chegam ao warehouse. Você pode configurar isso no conector de PostgreSQL.