Garantias de processamento em oleodutos Lakeflow

Retentativas e repetições são inevitáveis em qualquer pipeline real, por isso esta página explica as garantias de processamento que os pipelines Lakeflow oferecem e como manter as partes que escreves em segurança para serem reexecutadas.

Overview

Duas propriedades relacionadas determinam se a re-execução de um pipeline é segura:

  • Idempotência significa que um pipeline produz o mesmo resultado independentemente de quantas vezes 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 registo afeta o resultado. Pelo menos uma vez o processamento garante que todos os registos são processados, mas uma falha e uma tentativa novamente podem processar alguns registos mais do que uma vez, o que pode causar duplicados. O processamento exatamente uma vez garante que cada registo afeta o resultado como se tivesse sido processado precisamente uma vez, mesmo entre tentativas, sem duplicados nem lacunas.

Os pipelines do Lakeflow são idempotentes por defeito para os componentes que gerem e proporcionam processamento exatamente uma vez nas respetivas tabelas geridas. O importante é perceber onde essas garantias deixam de ser automáticas, para que possa adicionar os mecanismos de proteção certos nas fronteiras do seu pipeline.

Como funciona

Os pipelines do Lakeflow fornecem processamento exatamente uma vez e idempotência para os fluxos que gerem, e fornecem-lhe também ferramentas para garantir a idempotência da lógica que escreve.

Processamento único para tabelas geridas

Nas tabelas geridas, tem processamento exatamente uma vez por defeito. As tabelas de streaming utilizam pontos de controlo do Structured Streaming combinados com as escritas transacionais do Delta Lake: cada micro-lote confirma em conjunto os offsets da origem e os resultados produzidos, pelo que um micro-lote repetido após uma falha ou é concluído com êxito na totalidade, ou é integralmente revertido e repetido, nunca sendo parcialmente aplicado duas vezes. Isto aplica-se à ingestão de ficheiros com o Auto Loader, às operações de leitura do Kafka, do Kinesis e do Hubs de Eventos do Azure, e a AUTO CDC upserts, sem ter de escrever qualquer código.

Se uma fonte pelo menos uma vez enviar o mesmo registo várias vezes, o pipeline processa-os como registos únicos e escreve todos na tua tabela. Remover esses duplicados é da tua responsabilidade. Veja Desduplicar fontes pelo menos uma vez.

A idempotência nas leituras decorre desses mesmos pontos de verificação. O Auto Loader e os checkpoints de tabela de streaming garantem que cada ficheiro fonte ou offset é processado uma vez para efeitos de rastreamento de estados, pelo que o reprocessamento de uma atualização do pipeline após uma falha recomeça a partir do checkpoint, em vez de reprocessar ou saltar dados. Consegue-se isto utilizando tabelas em fluxo sobre spark.readStream, em vez de ciclos de processamento em lote implementados manualmente. Consulte Tabelas de streaming.

Utilize o AUTO CDC em vez do MERGE escrito manualmente

AUTO CDC INTO é inerentemente idempotente em relação ao seu keys e sequence_by. Aplicar o mesmo registo de alteração duas vezes, ou aplicar registos 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 recente do que o que já está armazenado:

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 escreveres a tua própria lógica de upsert fora de AUTO CDC (raro, mas por vezes necessário para condições de fusão complexas), baseia-a numa chave de negócio estável e garante que é segura se for aplicada duas vezes, por exemplo, uma MERGE ... WHEN MATCHED baseada em order_id em vez de uma INSERT cega. Para mais informações, consulte As APIs AUTO CDC: Simplificar a captura de dados de alterações com pipelines.

Mantém as tuas 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 vistas materializadas. Como uma vista materializada pode recomputar total ou incrementalmente, evite funções cuja saída depende do momento em que executam 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 ser escrito; obtenha a marca temporal do evento de origem ou passe-a como parâmetro para que a recomputação produza um resultado idêntico.
  • Conceba atualizações integrais por segurança. Uma atualização completa elimina e recalcula uma tabela do zero, o que só é seguro se todas as fontes de origem ainda conseguirem fornecer o histórico completo. Se uma fonte a montante apenas expõe uma janela deslizante de alterações, uma atualização completa de uma tabela AUTO CDC a jusante pode fazer perder silenciosamente o histórico, por isso conceba a retenção da fonte e do tópico tendo isso em conta.

Vai exatamente uma vez nas extremidades

É nos limites daquilo que o pipeline controla diretamente, como as escritas em sistemas externos, que o exactly-once deixa de ser automático. Quando te expandes para um sistema externo, faz com que a própria escrita seja idempotente, por exemplo, fazendo upserting por chave no lado receptor, já que um micro-batch retentado poderia escrever o mesmo lote duas vezes. O destino seguinte escreve cada partição do lote nos executores e utiliza uma chave de idempotência para que um lote repetido não seja escrito duas vezes:

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 a gravação em sistemas externos, consulte Sinks in Lakeflow pipelines.

Deduplicar fontes pelo menos uma vez

Quando uma fonte pode entregar um registo mais do que uma vez, elimine os duplicados a jusante. Combine uma marca-d'água com dropDuplicatesWithinWatermark, que é compatível com marcas-d'água e não requer um estado ilimitado para detetar duplicados. Eliminar duplicados com base nas colunas que identificam de forma única um evento. A identidade pode abranger várias colunas quando nenhuma coluna única é única por si só. No exemplo seguinte, um número de sequência de cliques é único apenas dentro da sua sessão, pelo que 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 unicidade da fonte, não do que parece distinto nos dados de amostra. Colunas que podem repetir legitimamente descartam eventos reais quando os tratamos como identidade. Um utilizador que clica duas vezes no mesmo anúncio é um exemplo comum: desduplicar com base no utilizador e no anúncio descarta silenciosamente o segundo clique.

AUTO CDCA semântica upsert baseada em chaves também colapsa duplicados naturalmente, por isso encaminhar pelo menos uma vez dados através de um AUTO CDC fluxo com chave de negócio estável é outra forma de convergir para um estado exatamente único.

Limitações

O processamento exatamente uma vez aplica-se a fluxos geridos Delta-para-Delta. Trate as seguintes arestas como pelo menos uma vez e acrescente lógica explícita de deduplicação ou escrita de idempotentes:

  • foreach_batch_sink e escritas externas personalizadas. O Spark garante que um lote é processado pelo menos uma vez, mas um lote novamente processado após uma gravação parcial pode fazer com que algumas linhas fiquem visíveis duas vezes no sistema externo. Assegure que a escrita externa é idempotente, por exemplo, fazendo uma inserção ou atualização com base numa chave natural ou escrevendo um ID de lote com base no qual o recetor possa eliminar duplicados.
  • Kafka como um lava-loiça. Os tópicos Kafka não suportam escritas transacionais exatamente uma vez como o Delta, por isso uma escrita micro-batch retentada para Kafka pode produzir mensagens duplicadas. Se os consumidores a jusante forem sensíveis a duplicados, elimine os duplicados do lado do consumidor, por exemplo, com base no ID do evento.
  • Fontes de dados personalizadas em Python usadas como fontes. O facto de as leituras serem processadas exatamente uma vez depende de a implementação da origem comunicar corretamente os offsets e retomar a partir deles. Se não registar os deslocamentos, trate-o pelo menos uma vez e desdupe a jusante com dropDuplicates num ID de evento ou confiando na AUTO CDCsemântica upsert baseada em chaves de .

Como regra geral, se todo o teu pipeline for Delta-para-Delta (tabelas de streaming e vistas materializadas a ler e escrever tabelas Delta através de fluxos geridos), já tens exatamente uma vez. Assim que adicionar um foreach_batch_sink, um destino não Delta ou uma origem personalizada não verificada, trate essa ligação específica como sendo at-least-once e adicione, nesse ponto, lógica de escrita idempotente ou de eliminação de duplicados.

Recursos adicionais

  • Last updated on