Garantias de processamento nos pipelines do Lakeflow

Repetições e reexecuções são inevitáveis em qualquer pipeline real, por isso esta página explica as garantias de processamento que os pipelines do Lakeflow oferecem e como manter para reexecução os componentes que você implementa.

Overview

Duas propriedades relacionadas determinam se executar novamente um pipeline é seguro:

  • Idempotência significa que um pipeline produz o mesmo resultado, não importa quantas vezes você o execute sobre a mesma entrada. Reexecutar após uma falha, preencher um intervalo de datas duas vezes ou reiniciar manualmente um trabalho nunca cria linhas duplicadas ou corrompe o estado.
  • A garantia de processamento descreve quantas vezes cada registro afeta o resultado. O processamento pelo menos uma vez garante que todos os registros sejam processados, mas uma falha e uma nova tentativa podem fazer com que alguns registros sejam processados mais de uma vez, o que traz o risco de duplicação. O processamento exatamente uma vez garante que cada registro afete o resultado como se tivesse sido processado precisamente uma única vez, mesmo após novas tentativas, sem duplicatas nem lacunas.

Os pipelines do Lakeflow são idempotentes por padrão para os componentes que gerenciam e oferecem processamento exatamente uma vez em suas próprias tabelas gerenciadas. O importante é entender onde essas garantias deixam de ser automáticas, para que você possa adicionar as salvaguardas certas nas bordas do seu pipeline.

Como funciona

Os pipelines do Lakeflow oferecem processamento exatamente uma vez e idempotência para os fluxos que gerenciam, além de ferramentas para manter idempotente também a lógica que você implementa.

Processamento único para tabelas gerenciadas

Em tabelas gerenciadas, há processamento exatamente uma vez por padrão. Tabelas de streaming usam checkpoints do Structured Streaming combinados com as gravações transacionais do Delta Lake: cada microlote confirma em conjunto seus deslocamentos de origem e sua saída, de modo que um lote reprocessado após uma falha ou é concluído com êxito por completo, ou é totalmente revertido e reprocessado, sem nunca ser aplicado parcialmente duas vezes. Isso vale para a ingestão de arquivos do Carregador Automático, leituras de Kafka, Kinesis e Hubs de Eventos do Azure, e upserts AUTO CDC, sem que você precise escrever nenhum código.

Se uma fonte pelo menos uma vez enviar o mesmo registro várias vezes, o pipeline os processa como registros únicos e grava todos na sua tabela. Remover esses duplicados é sua responsabilidade. Consulte Desduplicar fontes pelo menos uma vez.

A idempotência das leituras decorre desses mesmos checkpoints. O Carregador Automático e os checkpoints de tabelas de streaming garantem que cada arquivo de origem ou deslocamento seja processado uma única vez para rastreamento de estado, de modo que o reprocessamento de uma atualização do pipeline após uma falha seja retomado a partir do checkpoint, em vez de reprocessar ou ignorar dados. Você obtém isso usando tabelas de streaming sobre spark.readStream em vez de loops em lote implementados manualmente. Consulte tabelas de streaming.

Use o AUTO CDC em vez do MERGE escrito manualmente

AUTO CDC INTO é inerentemente idempotente em relação a seu keys e sequence_by. Aplicar o mesmo registro de alteração duas vezes, ou aplicar registros fora de ordem, resulta no mesmo estado final, porque o pipeline usa a coluna de sequência para decidir se uma linha recebida é realmente mais nova do que a que está armazenada:

CREATE FLOW customers_cdc_flow AS AUTO CDC INTO customers_silver
FROM stream(customers_cdc_bronze)
KEYS (customer_id)
SEQUENCE BY sequence_num
STORED AS SCD TYPE 1;

Se você escrever sua própria lógica de upsert fora de AUTO CDC (raro, mas às vezes necessário para condições complexas de mesclagem), baseie-a em uma chave de negócio estável e garanta que ela possa ser aplicada duas vezes com segurança, por exemplo, um MERGE ... WHEN MATCHED baseado em order_id, em vez de um INSERT sem critério. Para mais informações, consulte As APIs AUTO CDC: simplifique a captura de alterações de dados usando pipelines.

Mantenha suas próprias transformações idempotentes

Para manter a lógica idempotente ao reexecutar operações de escrita, siga estas duas diretrizes:

  • Evite transformações não determinísticas em visões materializadas. Como uma visualização materializada pode ser recomputada total ou incrementalmente, evite funções cuja saída depende de quando elas são executadas, e não do que é a entrada. Por exemplo, não use current_timestamp() para calcular um valor de negócio que deve permanecer fixo depois de registrado; obtenha o carimbo de data/hora do evento de origem ou passe-o como parâmetro para que a recomputação produza uma saída idêntica.
  • Projete atualizações completas por segurança. Uma atualização completa descarta e recalcula uma tabela do zero, o que só é seguro se toda fonte upstream ainda puder fornecer o histórico completo. Se uma origem upstream disponibiliza apenas uma janela contínua de alterações, uma atualização completa de uma tabela AUTO CDC downstream pode fazer com que o histórico seja perdido silenciosamente, portanto projete a retenção da origem e do tópico com isso em mente.

Obtenha exatamente uma vez nas bordas

É nos limites do que o pipeline controla diretamente, como as escritas em sistemas externos, que a garantia de execução exatamente uma vez deixa de ser automática. Quando você gravar em um sistema externo, faça com que a gravação em si seja idempotente, por exemplo, usando upsert pela chave no sistema de destino, pois um microlote reprocessado em uma nova tentativa poderia, por outro lado, gravar o mesmo lote duas vezes. O coletor a seguir grava cada partição do lote a partir dos executores e usa uma chave de idempotência para que um lote repetido não faça dupla gravação:

from pyspark import pipelines as dp

@dp.foreach_batch_sink(name="orders_to_external_api")
def write_orders_to_api(batch_df, batch_id):
    def write_partition(rows):
        # Open one client per partition.
        for row in rows:
            # Use an idempotency key (order_id) so a retried batch doesn't double-write.
            upsert_to_external_system(key=row.order_id, payload=row.asDict())

    batch_df.select("order_id", "amount").foreachPartition(write_partition)

Para mais informações sobre como escrever em sistemas externos, consulte Coletores em pipelines do Lakeflow.

Desduplicar fontes pelo menos uma vez

Quando uma fonte puder entregar um registro mais de uma vez, elimine duplicatas nas etapas posteriores. Combine uma marca-d’água com dropDuplicatesWithinWatermark, que é compatível com marcas-d’água e não exige estado não limitado para detectar duplicatas. Elimine duplicatas com base nas colunas que identificam exclusivamente um evento. A identidade pode abranger várias colunas quando nenhuma coluna única é única por si só. No exemplo a seguir, um número de sequência de cliques é único apenas dentro da sua sessão, então as duas colunas juntas identificam o evento:

from pyspark import pipelines as dp

@dp.table(name="clicks_deduped")
def clicks_deduped():
    return (
        spark.readStream.table("clicks_bronze")
        .withWatermark("click_ts", "5 minutes")
        .dropDuplicatesWithinWatermark(["session_id", "click_seq_num"])
    )

Escolha essas colunas do contrato de exclusividade da fonte, não do que parece distinto nos dados de amostra. Colunas que podem repetir legitimamente descartam eventos reais quando você os trata como identidade. Um usuário clicando duas vezes no mesmo anúncio é um exemplo comum: ao desduplicar o usuário, o anúncio descarta silenciosamente o segundo clique.

A semântica de upsert baseada em chave de AUTO CDC também recolhe duplicatas naturalmente, então encaminhar dados pelo menos uma vez por meio de um fluxo AUTO CDC indexado por uma chave de negócio estável é outra forma de convergir para um estado de exatamente uma vez.

Limitações

O processamento exatamente uma vez se aplica aos fluxos gerenciados Delta para Delta. Trate as seguintes bordas como pelo menos uma vez e adicione lógica explícita de desduplicação ou gravação idempotente ali:

  • foreach_batch_sink e gravações externas personalizadas. O Spark garante que um lote seja tentado pelo menos uma vez, mas um lote retentado após uma gravação parcial pode deixar algumas linhas visíveis duas vezes no sistema externo. Torne a gravação externa idempotente, por exemplo, usando upsert com base em uma chave natural ou gravando um ID de lote que o receptor possa usar para eliminar duplicatas.
  • Kafka como coletor. Tópicos de Kafka não suportam gravações transacionais exatamente uma vez como o Delta, então uma gravação de microlote repetida no Kafka pode produzir mensagens duplicadas. Se os consumidores downstream forem sensíveis a duplicatas, faça a desduplicação no lado do consumidor, por exemplo, pelo ID do evento.
  • Fontes de dados personalizadas em Python usadas como fontes. Se as leituras ocorrem exatamente uma vez, isso depende de a implementação da sua origem relatar e retomar corretamente a partir dos deslocamentos. Se não rastrear deslocamentos, trate-o como pelo menos uma vez e faça a desduplicação downstream com dropDuplicates com base em um ID de evento ou confiando na semântica de upsert baseada em chave de AUTO CDC.

Como regra prática, se todo o seu pipeline for de Delta para Delta (tabelas de streaming e exibições materializadas lendo e gravando tabelas Delta por meio de fluxos gerenciados), você já conta com processamento exatamente uma vez. No momento em que você adicionar um foreach_batch_sink, um coletor não Delta, ou uma fonte personalizada não verificada, trate essa borda específica como pelo menos uma vez e adicione lógica de gravação idempotente ou desduplicação ali.

Recursos adicionais

  • Last updated on