O que é o acompanhamento de progresso assíncrono?

O acompanhamento de progresso assíncrono permite que os pipelines de Fluxo estruturado verifiquem o progresso de forma assíncrona e paralelamente ao processamento de dados real em um microlote, reduzindo a latência associada à manutenção de offsetLog e commitLog.

Nota

O acompanhamento de progresso assíncrono não funciona com gatilhos Trigger.once ou Trigger.availableNow. A tentativa de habilitar esse recurso com esses gatilhos resulta em falha de consulta.

Como o acompanhamento de progresso assíncrono funciona para reduzir a latência?

O Structured Streaming baseia-se na persistência e no gerenciamento de offsets como indicadores de progresso para o processamento de consultas. A operação de gerenciamento de offset afeta diretamente a latência de processamento, uma vez que nenhum processamento de dados pode ocorrer até que essas operações sejam concluídas. O acompanhamento de progresso assíncrono permite que os pipelines de Streaming Estruturado registrem o progresso sem serem afetados por essas operações de gerenciamento de offsets.

Quando você deve configurar a frequência de ponto de verificação?

Os usuários podem configurar a frequência com que o progresso está em ponto de verificação. As configurações padrão para frequência de ponto de verificação fornecem uma boa taxa de transferência para a maioria das consultas. Configurar a frequência é útil para cenários em que as operações de gerenciamento de deslocamento ocorrem a uma taxa mais alta do que podem ser processadas, o que cria um acúmulo cada vez maior de operações de gerenciamento de deslocamento. Para conter essa lista de pendências crescente, o processamento de dados é bloqueado ou lento, essencialmente revertendo o comportamento de processamento para eliminar os benefícios do acompanhamento de progresso assíncrono.

Nota

O tempo de recuperação de falha aumenta com o aumento no tempo do intervalo de ponto de verificação. Em caso de falha, um pipeline precisa reprocessar todos os dados antes do ponto de verificação bem-sucedido anterior. Os usuários podem considerar essa compensação entre latência mais baixa durante o processamento regular e o tempo de recuperação em caso de falha.

Quais configurações estão associadas ao acompanhamento de progresso assíncrono?

Opção	Valor	Padrão	Descrição
rastreamentoDeProgressoAssincronoAtivado	verdadeiro/falso	falso	habilitar ou desabilitar o acompanhamento de progresso assíncrono
asyncProgressTrackingCheckpointIntervalMs	Milissegundos	1000	o intervalo no qual confirmamos deslocamentos e confirmações de conclusão

Como os usuários podem habilitar o acompanhamento de progresso assíncrono?

Os usuários podem usar código semelhante ao código abaixo para habilitar esse recurso:

val stream = spark.readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
      .option("subscribe", "in")
      .load()

val query = stream.writeStream
     .format("kafka")
        .option("topic", "out")
     .option("checkpointLocation", "/tmp/checkpoint")
        .option("asyncProgressTrackingEnabled", "true")
     .start()

Desativar o acompanhamento de progresso assíncrono

Quando o acompanhamento de progresso assíncrono está habilitado, a estrutura não verifica o progresso de cada lote. Para resolver isso, antes de desabilitar o acompanhamento de progresso assíncrono, processe pelo menos dois microlotes com as seguintes configurações:

• .option("asyncProgressTrackingEnabled", "true")
• .option("asyncProgressTrackingCheckpointIntervalMs", 0)

Interrompa a consulta depois que pelo menos dois microlotes tiverem concluído o processamento. Agora você pode desabilitar com segurança o acompanhamento de progresso assíncrono e reiniciar a consulta.

Se você tiver desabilitado o acompanhamento de progresso assíncrono sem concluir esta etapa, poderá encontrar o seguinte erro:

java.lang.IllegalStateException: batch x doesn't exist

Nos logs de driver, você pode ver o seguinte erro:

The offset log for batch x doesn't exist, which is required to restart the query from the latest batch x from the offset log. Please ensure there are two subsequent offset logs available for the latest batch via manually deleting the offset file(s). Please also ensure the latest batch for commit log is equal or one batch earlier than the latest batch for offset log.

Seguir as instruções nesta seção para desabilitar o acompanhamento de progresso assíncrono permite que você resolva esses erros e repare sua carga de trabalho de streaming.

Limitações com acompanhamento de progresso assíncrono

Esse recurso tem as seguintes limitações:

• O acompanhamento de progresso assíncrono só tem suporte em pipelines sem estado ao usar o Kafka como coletor.
• Exatamente uma vez, o processamento de ponta a ponta não é garantido com rastreamento de progresso assíncrono, porque os intervalos de deslocamento para lote podem ser alterados em caso de falha. Alguns coletores, como Kafka, nunca fornecem garantias de exatamente uma vez.