Pular para o conteúdo principal
Última atualização em

Garantias de processamento em LakeFlow Pipelines

Em qualquer pipeline real, novas tentativas e execuções são inevitáveis. Por isso, esta página explica as garantias de processamento que os LakeFlow Pipelines oferecem e como manter as partes que você escreve seguras para serem executadas novamente.

Visão geral​

Duas propriedades relacionadas determinam se a reexecução de um pipeline é segura:

  • Idempotência significa que um pipeline produz o mesmo resultado, não importa quantas vezes você o execute sobre a mesma entrada. Executar novamente após uma falha, preencher retroativamente um intervalo de datas duas vezes ou reacionar manualmente um job nunca cria linhas duplicadas nem corrompe o estado.
  • Garantia de processamento descreve quantas vezes cada registro afeta o resultado. O processamento pelo menos uma vez garante que cada registro seja processado, mas uma falha e uma nova tentativa podem processar alguns registros mais de uma vez, o que gera risco de duplicatas. O processamento exatamente uma vez garante que cada registro afete o resultado como se tivesse sido processado precisamente uma vez, mesmo em novas tentativas, sem duplicatas e sem lacunas.

Os LakeFlow Pipelines são idempotentes por default para as partes que gerenciam e oferecem processamento exatamente uma vez dentro de suas próprias tabelas gerenciadas. O importante é entender onde essas garantias deixam de ser automáticas, para que você possa adicionar as salvaguardas corretas nas bordas do seu pipeline.

Como funciona​

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

Processamento exatamente uma vez para tabelas gerenciadas​

Em tabelas gerenciadas, o processamento é realizado exatamente uma vez por default. As tabelas de transmissão usam pontos de verificação do Structured Streaming combinados com as gravações transacionais do Delta Lake: cada microlote commit seus offsets de origem e sua saída juntos, de modo que um dos lotes repetido após uma falha seja totalmente bem-sucedido ou seja totalmente revertido e repetido, nunca aplicado parcialmente duas vezes. Isso vale para ingestão automática de arquivos do Auto Loader, leituras de Kafka, Kinesis e Azure Event Hubs e atualizações de AUTO CDC, sem nenhum código de sua parte.

Se uma fonte de dados com capacidade de envio mínimo enviar o mesmo registro várias vezes, o pipeline os processará como registros únicos e os gravará todos na sua tabela. A remoção desses itens duplicados é de sua responsabilidade. Consulte Remover fontes duplicadas pelo menos uma vez.

A idempotência para leituras decorre desses mesmos pontos de verificação. Os pontos de verificação do Auto Loader e da tabela de transmissão garantem que cada arquivo de origem ou deslocamento seja processado uma vez para fins de acompanhamento de estado, portanto, o reprocessamento de uma atualização de pipeline após uma falha é retomado a partir do ponto de verificação, em vez de reprocessar ou ignorar dados. Você obtém isso usando tabelas de transmissão sobre spark.readStream em vez de loops de lotes manuais. Consulte Tabelas de transmissão.

Use o AUTO CDC em vez de MERGE escrito manualmente​

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

SQL
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 merge), use uma key de negócio estável e torne-a segura para ser aplicada duas vezes, por exemplo, um MERGE ... WHEN MATCHED com key em order_id em vez de um INSERT cego. Para obter mais informações, consulte As APIs de CDC AUTOMÁTICO: simplifique a captura de dados de alterações (CDC) com pipelines.

Mantenha suas próprias transformações idempotentes​

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

  • Evite transformações não determinísticas em visões materializadas. Como uma visão materializada pode recalcular total ou incrementalmente, evite funções cuja saída dependa de quando elas são de execução, em vez de qual é a entrada. Por exemplo, não use current_timestamp() para calcular um valor comercial que deve permanecer fixo após ser gravado; obtenha o timestamp do evento de origem ou passe-o como um parâmetro para que o recálculo produza uma saída idêntica.
  • Projete refresh completos para serem seguros. Um refresh completo descarta e recalcula uma tabela do zero, o que só é seguro se cada fonte upstream ainda puder produzir o histórico completo. Se uma fonte upstream expõe apenas uma janela móvel de alterações, um refresh completo de uma tabela AUTO CDC downstream pode perder a história silenciosamente; portanto, projete a retenção de fonte e tópico com isso em mente.

Obter semântica de processamento exatamente uma vez nas bordas​

Onde o processamento exactly-once deixa de ser automático é nas bordas do que o pipeline controla diretamente, como gravações em sistemas externos. Ao distribuir para um sistema externo, torne a própria gravação idempotente, por exemplo, fazendo upsert por key no lado receptor, já que um micro-batch repetido poderia, de outra forma, gravar o mesmo lote duas vezes. O coletor a seguir grava cada partição do lote a partir do executor e usa uma key de idempotência para que um lote repetido não seja gravado duas vezes:

Python
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 obter mais informações sobre como gravar em sistemas externos, consulte Sinks em LakeFlow Pipelines.

Desduplicar fontes de pelo menos uma vez​

Quando uma fonte pode entregar um registro mais de uma vez, deduplique downstream. Combine uma marca d'água com dropDuplicatesWithinWatermark, que reconhece marcas d'água e não requer estado ilimitado para detectar duplicatas. Deduplique nas colunas que identificam exclusivamente um evento. A identidade pode abranger várias colunas quando nenhuma coluna única é exclusiva por si só. No exemplo a seguir, um número de sequência de clique é exclusivo apenas dentro de sua sessão, portanto, as duas colunas juntas identificam o evento:

Python
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 a partir do contrato de exclusividade da origem, não do que parece distinto nos dados de amostra. Colunas que podem se repetir legitimamente descartam eventos reais quando você as trata como identidade. Um usuário clicando no mesmo anúncio duas vezes é um exemplo comum: a desduplicação no usuário e no anúncio descarta silenciosamente o segundo clique.

AUTO CDCAs semânticas de upsert baseadas em key também colapsam duplicatas naturalmente, portanto, rotear dados de "pelo menos uma vez" através de um fluxo AUTO CDC baseado em uma key de negócio estável é outra maneira de convergir para um estado de "exatamente uma vez".

Limitações​

O processamento exactly-once aplica-se a fluxos Delta-to-Delta gerenciados. Trate as seguintes bordas como at-least-once e adicione lógica explícita de desduplicação ou de gravação idempotente nelas:

  • foreach_batch_sink e gravações externas personalizadas. O Spark garante que uma tentativa de lote seja feita pelo menos uma vez, mas um lote reexecutado 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, fazendo upsert em uma chave natural ou gravando um ID de lote no qual o receptor possa realizar a desduplicação.
  • Kafka como um sink. Os tópicos do Kafka não oferecem suporte a gravações transacionais de "exatamente uma vez" da mesma forma que o Delta, portanto, um micro-lote reexecutado gravando no Kafka pode produzir mensagens duplicadas. Se os consumidores downstream forem sensíveis a duplicidades, faça a desduplicação no lado do consumidor, por exemplo, por ID de evento.
  • Fontes de dados Python personalizadas usadas como fontes. Se as leituras são exatamente uma vez depende se a implementação da sua fonte relata e retoma corretamente a partir dos offsets. Se não rastrear offsets, trate como pelo menos uma vez e deduplique a jusante com dropDuplicates em um ID de evento ou confiando na semântica de upsert baseada em key do AUTO CDC.

Como regra geral, se todo o seu pipeline for Delta-para-Delta (tabelas de transmissão e visualizações materializadas lendo e gravando tabelas Delta por meio de fluxos gerenciados), você já tem a garantia de "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 aresta específica como "pelo menos uma vez" e adicione lógica de gravação idempotente ou de deduplicação ali.

Outros recursos​

Nesta página