Exatamente uma vez no Dataflow

O Dataflow suporta o processamento de registos exatamente uma vez. Esta página explica como o Dataflow implementa o processamento exatamente uma vez, ao mesmo tempo que garante uma baixa latência.

Vista geral

Os pipelines em lote usam sempre o processamento exatamente uma vez. Os pipelines de streaming usam o processamento exatamente uma vez por predefinição, mas também podem usar o processamento pelo menos uma vez.

O processamento exatamente uma vez oferece garantias sobre os resultados do processamento de registos, incluindo os resultados de cada fase do pipeline. Especificamente, para cada registo que chega ao pipeline a partir de uma origem ou chega a uma fase a partir de uma fase anterior, o Dataflow garante o seguinte:

  • O registo é processado e não é perdido.
  • Quaisquer resultados do processamento que permaneçam no pipeline são refletidos, no máximo, uma vez.

Por outras palavras, os registos são processados, pelo menos, uma vez, e os resultados são confirmados exatamente uma vez.

O processamento exatamente uma vez garante que os resultados são precisos, sem registos duplicados na saída. O Dataflow está otimizado para minimizar a latência, mantendo a semântica exatamente uma vez. No entanto, o processamento exatamente uma vez continua a incorrer em custos para realizar a remoção de duplicados. Para exemplos de utilização que podem tolerar registos duplicados, pode muitas vezes reduzir o custo e melhorar a latência ativando o modo "pelo menos uma vez". Para mais informações sobre a escolha entre o streaming exatamente uma vez e, pelo menos, uma vez, consulte o artigo Defina o modo de streaming do pipeline.

Dados em atraso

O processamento exatamente uma vez garante a precisão do pipeline: se o pipeline processar um registo, o Dataflow garante que o registo é refletido na saída e que o registo não é duplicado.

No entanto, numa pipeline de streaming, o processamento exatamente uma vez não pode garantir que os resultados estão concluídos, porque os registos podem chegar tarde. Por exemplo, suponha que o seu pipeline executa uma agregação num período, como Count. Com o processamento exatamente uma vez, o resultado é preciso para os registos que chegam dentro do período de tempo de forma atempada, mas os registos tardios podem ser ignorados.

Geralmente, não existe forma de garantir a integridade num pipeline de streaming, porque, em teoria, os registos podem chegar arbitrariamente tarde. No caso limite, teria de esperar eternamente para produzir um resultado. De forma mais prática, o Apache Beam permite-lhe configurar o limite para rejeitar dados atrasados e quando emitir resultados agregados. Para mais informações, consulte o artigo Marcas de água e dados tardios na documentação do Apache Beam.

Efeitos secundários

Não é garantido que os efeitos secundários tenham uma semântica exatamente uma vez. É importante notar que isto inclui a escrita de resultados num armazenamento externo, a menos que o destino também implemente uma semântica exatamente uma vez.

Especificamente, o Dataflow não garante que cada registo passe por cada transformação exatamente uma vez. Devido a novas tentativas ou falhas de trabalhadores, o Dataflow pode enviar um registo através de uma transformação várias vezes ou até em simultâneo em vários trabalhadores.

Como parte do processamento exatamente uma vez, o Dataflow remove as duplicados das saídas. No entanto, se o código numa transformação tiver efeitos secundários, esses efeitos podem ocorrer várias vezes. Por exemplo, se uma transformação fizer uma chamada de serviço remoto, essa chamada pode ser feita várias vezes para o mesmo registo. Os efeitos secundários podem até levar à perda de dados em algumas situações. Por exemplo, suponhamos que uma transformação lê um ficheiro para produzir resultados e, em seguida, elimina imediatamente o ficheiro sem esperar que os resultados sejam confirmados. Se ocorrer um erro ao confirmar o resultado, o Dataflow tenta novamente a transformação, mas agora a transformação não consegue ler o ficheiro eliminado.

Registo

O resultado do registo do processamento indica que o processamento ocorreu, mas não indica se os dados foram confirmados. Por conseguinte, os ficheiros de registo podem indicar que os dados foram processados várias vezes, embora os resultados dos dados processados sejam confirmados no armazenamento persistente apenas uma vez. Além disso, os registos nem sempre refletem os dados processados e confirmados. Os registos podem ser ignorados devido à limitação ou perdidos devido a outros problemas do serviço de registo.

Streaming exatamente uma vez

Esta secção explica como o Dataflow implementa o processamento exatamente uma vez para tarefas de streaming, incluindo a forma como o Dataflow gere complexidades como o processamento não determinístico, os dados tardios e o código personalizado.

Dataflow streaming shuffle

As tarefas de streaming do Dataflow são executadas em muitos trabalhadores diferentes em paralelo, atribuindo intervalos de trabalho a cada trabalhador. Embora as atribuições possam mudar ao longo do tempo em resposta a falhas de trabalhadores, ao dimensionamento automático ou a outros eventos, após cada transformação GroupByKey, todos os registos com a mesma chave são processados no mesmo trabalhador. A transformação GroupByKey é frequentemente usada por transformações compostas, como Count, FileIO e assim sucessivamente. Para garantir que os registos de uma determinada chave acabam no mesmo trabalhador, os trabalhadores do Dataflow misturam os dados entre si através de chamadas de procedimentos remotos (RPCs).

Para garantir que os registos não são perdidos durante a ordenação aleatória, o Dataflow usa uma cópia de segurança a montante. Com a cópia de segurança a montante, o trabalhador que envia os registos tenta novamente os RPCs até receber uma confirmação positiva de que o registo foi recebido. Os efeitos secundários do processamento do registo são confirmados no armazenamento persistente a jusante. Se o trabalhador que envia os registos ficar indisponível, o Dataflow continua a tentar novamente as RPCs, o que garante que todos os registos são entregues, pelo menos, uma vez.

Uma vez que estas novas tentativas podem criar duplicados, todas as mensagens são etiquetadas com um ID único. Cada recetor armazena um catálogo de todos os IDs que já foram vistos e processados. Quando é recebido um registo, o Dataflow procura o respetivo ID no catálogo. Se o ID for encontrado, o registo já foi recebido e confirmado, e é ignorado como duplicado. Para garantir que os IDs dos registos são estáveis, cada resultado de passo a passo é verificado no armazenamento. Como resultado, se a mesma mensagem for enviada várias vezes devido a chamadas RPC repetidas, a mensagem só é confirmada no armazenamento uma vez.

Garantir uma latência baixa

Para que o processamento exatamente uma vez seja viável, a E/S tem de ser reduzida, em particular, impedindo a E/S em todos os registos. Para atingir este objetivo, o Dataflow usa filtros de Bloom e recolha de lixo.

Filtros de brilho

Os filtros de Bloom são estruturas de dados compactas que permitem verificações rápidas de associação a conjuntos. No Dataflow, cada trabalhador mantém um filtro de Bloom de todos os IDs que vê. Quando chega um novo ID de registo, o trabalhador procura o ID no filtro. Se o filtro devolver o valor falso, este registo não é um duplicado e o trabalhador não procura o ID no armazenamento estável.

O Dataflow mantém um conjunto de filtros de Bloom contínuos agrupados por tempo. Quando um registo chega, o Dataflow escolhe o filtro adequado para verificar com base na data/hora do sistema. Este passo impede a saturação dos filtros de Bloom à medida que os filtros são recolhidos como lixo e também limita a quantidade de dados que têm de ser analisados no arranque.

Recolha de lixo

Para evitar o preenchimento do armazenamento com IDs de registos, o Dataflow usa a recolha de lixo para remover registos antigos. O Dataflow usa a data/hora do sistema para calcular uma marca d'água de recolha de lixo.

Esta marca de água baseia-se na quantidade de tempo físico gasto à espera numa determinada fase. Por conseguinte, também fornece informações sobre que partes do pipeline são lentas. Estes metadados são a base para a métrica de atraso do sistema apresentada na interface de monitorização do Dataflow.

Se um registo chegar com uma data/hora anterior à marca d'água e os IDs desta data/hora já tiverem sido recolhidos, o registo é ignorado. Uma vez que a marca d'água baixa que aciona a recolha de lixo não avança até que as entregas de registos sejam confirmadas, estes registos que chegam tarde são duplicados.

Fontes não determinísticas

O Dataflow usa o SDK do Apache Beam para ler dados em pipelines. Se o processamento falhar, o Dataflow pode tentar novamente as leituras a partir de uma origem. Nessa situação, o Dataflow tem de garantir que cada registo único produzido por uma origem é registado exatamente uma vez. Para origens determinísticas, como o Pub/Sub Lite ou o Kafka, os registos são lidos com base num desvio registado, o que mitiga a necessidade deste passo.

Uma vez que o Dataflow não pode atribuir automaticamente IDs de registos, as origens não determinísticas têm de indicar ao Dataflow quais são os IDs de registos para evitar a duplicação. Quando uma origem fornece IDs únicos para cada registo, o conector usa uma mistura no pipeline para remover duplicados. Os registos com o mesmo ID são filtrados. Para ver um exemplo de como o Dataflow implementa o processamento exatamente uma vez quando usa o Pub/Sub como origem, consulte a secção Processamento exatamente uma vez na página Streaming com o Pub/Sub.

Quando executa DoFns personalizados como parte do seu pipeline, o Dataflow não garante que este código seja executado apenas uma vez por registo. Para garantir o processamento pelo menos uma vez em caso de falhas do trabalhador, o Dataflow pode executar um determinado registo através de uma transformação várias vezes ou pode executar o mesmo registo em simultâneo em vários trabalhadores. Se incluir código no seu pipeline que execute ações como contactar um serviço externo, as ações podem ser executadas mais do que uma vez para um determinado registo.

Para tornar o processamento não determinístico efetivamente determinístico, use a criação de pontos de verificação. Quando usa a criação de pontos de verificação, cada resultado de uma transformação é criado como ponto de verificação no armazenamento estável com o respetivo ID exclusivo antes de ser enviado para a fase seguinte. As repetições na entrega aleatória do retransmissor do Dataflow enviam a saída que foi verificada. Embora o seu código possa ser executado várias vezes, o Dataflow garante que o resultado de apenas uma dessas execuções é armazenado. O Dataflow usa um armazenamento consistente que impede que sejam escritos duplicados no armazenamento estável.

Entrega de saída exatamente uma vez

O SDK Apache Beam inclui sinks incorporados concebidos para garantir que não produzem duplicados. Sempre que possível, use um destes lavatórios incorporados.

Se precisar de escrever o seu próprio destino, a melhor abordagem é tornar o objeto da função idempotente para que possa ser repetido com a frequência necessária sem causar efeitos secundários não intencionais. No entanto, muitas vezes, algum componente da transformação que implementa a funcionalidade do destino é não determinístico e pode mudar se for repetido.

Por exemplo, numa agregação em janela, o conjunto de registos na janela pode ser não determinístico. Especificamente, a janela pode tentar ser acionada com os elementos e0, e1 e e2. O trabalhador pode falhar antes de confirmar o processamento da janela, mas não antes de esses elementos serem enviados como efeito secundário. Quando o worker é reiniciado, a janela é acionada novamente e chega um elemento tardio e3. Uma vez que este elemento chega antes de a janela ser confirmada, não é contabilizado como dados tardios, pelo que a função DoFn é chamada novamente com os elementos e0, e1, e2 e e3. Estes elementos são, em seguida, enviados para a operação de efeito secundário. A idempotência não ajuda neste cenário, porque são enviados conjuntos de registos lógicos diferentes de cada vez.

Para resolver a não determinância no Dataflow, use a transformação Reshuffle integrada. Quando o Dataflow baralha os dados, escreve-os de forma duradoura para que todos os elementos gerados de forma não determinística sejam estáveis se as operações forem repetidas após o baralhamento. A utilização da transformação Reshuffle garante que apenas uma versão do resultado de um DoFn pode ultrapassar um limite de aleatorização. O seguinte padrão garante que a operação de efeito secundário recebe sempre um registo determinístico para a saída:

c.apply(Window.<..>into(FixedWindows.of(Duration.standardMinutes(1))))
 .apply(GroupByKey.<..>.create())
 .apply(new PrepareOutputData())
 .apply(Reshuffle.<..>of())
 .apply(WriteToSideEffect());

Para garantir que o executor do Dataflow sabe que os elementos têm de ser estáveis antes de executar um DoFn, adicione a anotação RequiresStableInput ao DoFn.

Saiba mais