Quero usar um acumulador para reunir algumas estatísticas sobre os dados que estou manipulando em um trabalho do Spark. Idealmente, eu faria isso enquanto o trabalho calcula as transformações necessárias, mas como o Spark recalcula tarefas em casos diferentes, os acumuladores não refletem métricas verdadeiras. Aqui está como a documentação descreve isso:

Para atualizações do acumulador executadas apenas em ações, o Spark garante que a atualização de cada tarefa no acumulador seja aplicada apenas uma vez, ou seja, tarefas reiniciadas não atualizarão o valor. Nas transformações, os usuários devem estar cientes de que a atualização de cada tarefa pode ser aplicada mais de uma vez se tarefas ou estágios do trabalho forem reexecutados.

Isso é confuso, já que a maioriaações não permitem a execução de código personalizado (onde acumuladores podem ser usados), eles geralmente obtêm os resultados de transformações anteriores (preguiçosamente). A documentação também mostra isso:

val acc = sc.accumulator(0)
data.map(x => acc += x; f(x))
// Here, acc is still 0 because no actions have cause the `map` to be computed.

Mas se adicionarmosdata.count() no final, isso seria garantido como correto (sem duplicatas) ou não? Claramenteacc não é usado "somente ações internas", pois o mapa é uma transformação. Portanto, não deve ser garantido.

Por outro lado, a discussão sobre tickets Jira relacionados fala sobre "tarefas de resultado" em vez de "ações". Por exemploaqui eaqui. Isso parece indicar que o resultado seria certamente garantido, pois estamos usandoacc imediatamente antes da ação e, portanto, deve ser computado como um único estágio.

Estou supondo que esse conceito de "tarefa de resultado" tenha a ver com o tipo de operações envolvidas, sendo o último que inclui uma ação, como neste exemplo, que mostra como várias operações são divididas em estágios (em magenta, imagem tirada deaqui):

Então, hipoteticamente, umcount() a ação no final dessa cadeia faria parte do mesmo estágio final, e eu estaria garantido que os acumuladores usados no último mapa não incluirão duplicatas?

Esclarecer esse problema seria ótimo! Obrigado.