Estamos gerando um índice seqüencial em um ParDo usando o Java SDK 2.0.0 da Beam. Assim como o exemplo simples de índice com estado em Beamintrodução ao processamento com estado nós usamos umValueState<Integer> cell e nossa única operação é recuperar o valor e incrementar quando precisamos do próximo índice:

Integer statefulIndex = firstNonNull(index.read(), 0);
index.write(statefulIndex + 1);

Ao executar com o Dataflow do Google, percebemos na interface de monitoramento do Dataflow que o tempo de parede desse ParDo estava acumulando em sincronia com o tempo decorrido. Conseguimos confirmar que o ParDo executa um thread único ssh'ing no nó do trabalhador e usandotop e1 para visualizar o uso da CPU por núcleo. Comentando a célula de processamento com estado e mantendo o código inalterado, o mesmo ParDo usa todos os núcleos de nossan1-standard-32 nó do trabalhador.

Mesmo que o executor do Dataflow consiga paralelizar a indexação com estado com base em cada par de chaves e janelas (atualmente temos uma janela e uma chave), a falta de paralelismo causa uma diminuição tão significativa no desempenho que não podemos usá-lo. Esse é o comportamento esperado do corredor do Dataflow?

Ingenuamente, eu esperava que, nos bastidores, a indexação estável da Beam operasse de maneira semelhante à da JavaAtomicInteger. Existem restrições que impedem o processamento paralelo com umValueState<Integer> célula ou essa funcionalidade ainda não está incorporada no corredor?