Realtime Compute for Apache Flink:Checkpoints e savepoints

• Por que os dados não são atualizados após a expiração do TTL de estado quando o mini-batch está ativado?

• Como calcular o horário de início do próximo checkpoint periódico

• Quais são as diferenças entre o GeminiStateBackend nas versões VVR 8.x e VVR 6.x?

• É normal um checkpoint completo e um incremental terem o mesmo tamanho?

• O que fazer se os checkpoints estiverem lentos em uma implantação Python

• Como solucionar exceções de checkpoint

• Erro: You are using the new V4 state engine to restore old state data from a checkpoint

• Erro: java.lang.NegativeArraySizeException

• Erro: org.apache.flink.streaming.connectors.kafka.FlinkKafkaException: Too many ongoing snapshots.

• Erro: Exceeded checkpoint tolerable failure threshold

Falhas na atualização de dados com mini-batch

O estado armazena os resultados de computações completas anteriores. Se o TTL do estado expirar, o sistema limpa esses resultados acumulados. Consequentemente, não é possível atualizar novos dados com base nos resultados do mini-batch.

Por outro lado, se o mini-batch estiver desativado, quando o TTL do estado expirar, o sistema recalcula e emite os dados da chave expirada. Isso garante atualizações contínuas de dados. No entanto, o aumento na frequência de atualizações pode causar outros problemas, como atrasos no processamento.

Portanto, configure as definições de mini-batch e TTL de acordo com seus requisitos de negócio específicos.

Cálculo do horário de início do próximo checkpoint

Dois parâmetros determinam o horário de início do próximo checkpoint: o intervalo de checkpoint e a pausa mínima entre checkpoints. O sistema aciona um novo checkpoint quando ambas as condições a seguir são atendidas:

• Intervalo de checkpoint: tempo mínimo entre o início de um checkpoint e o início do próximo. Corresponde ao período entre <start time of previous checkpoint, start time of next checkpoint>.

• Pausa mínima: tempo mínimo entre o fim de um checkpoint e o início do próximo. Refere-se ao intervalo entre <end time of previous checkpoint, start time of next checkpoint>.

Considere os dois cenários a seguir, em que o intervalo de checkpoint é de 3 minutos, a pausa mínima é de 3 minutos e o tempo limite é de 10 minutos.

• Cenário 1: A implantação executa normalmente e todos os checkpoints têm êxito.

  O primeiro checkpoint começa às 12:00:00 e é concluído com êxito às 12:00:02. O segundo checkpoint começará às 12:03:00.

• Cenário 2: A implantação apresenta comportamento anormal e um checkpoint falha devido a tempo limite.

  O primeiro checkpoint começa às 12:00:00 e é concluído com êxito às 12:00:02. O segundo checkpoint inicia às 12:03:00, mas falha às 12:13:00 por tempo limite. O terceiro checkpoint começará às 12:16:00.

Para obter mais informações sobre como configurar a pausa mínima entre checkpoints, consulte Tuning Checkpointing.

GeminiStateBackend no VVR 8.x vs. VVR 6.x

Por padrão, os mecanismos do Realtime Compute for Apache Flink para VVR 6.x usam o GeminiStateBackend V3, enquanto os mecanismos VVR 8.x utilizam a versão V4.

Checkpoints completos e incrementais com o mesmo tamanho

Se você observar que um checkpoint completo e um incremental possuem o mesmo tamanho, siga estas orientações:

• Verifique se o checkpoint incremental está configurado e ativado corretamente.

• Esse comportamento pode ser esperado em determinados cenários. Por exemplo:

  1. Antes da ingestão de dados (por exemplo, antes das 18:29), a implantação não processou nenhum dado. O checkpoint contém apenas o estado inicial da source, tornando-o efetivamente um checkpoint completo.

  2. Às 18:29, um milhão de registros são ingeridos. Se esses dados forem totalmente processados dentro do próximo intervalo de checkpoint (por exemplo, 3 minutos) e nenhum outro dado chegar, o primeiro checkpoint incremental conterá todo o estado gerado a partir desses registros.

  Nesse caso, espera-se que o checkpoint completo e o primeiro checkpoint incremental tenham o mesmo tamanho. O primeiro checkpoint incremental deve incluir o estado de todos os dados para garantir que uma recuperação completa seja possível a partir desse ponto, o que o torna funcionalmente equivalente a um checkpoint completo.

  Os benefícios do checkpoint incremental geralmente se tornam aparentes a partir do segundo checkpoint. Com uma entrada de dados estável e sem grandes alterações de estado, os checkpoints incrementais subsequentes devem ser menores, indicando que o sistema está criando snapshots corretamente apenas das alterações de estado. Se eles permanecerem com o mesmo tamanho, investigue o estado e o comportamento do seu sistema para identificar possíveis problemas.

Checkpoints lentos em implantações Python

• Causa

  Uma Função Definida pelo Usuário (UDF) Python com baixo desempenho pode aumentar a duração do checkpoint e degradar o desempenho da implantação.

• Solução

  Reduza o tamanho do buffer. Na seção Other Configuration, defina os seguintes parâmetros. Para obter instruções, consulte Configurar parâmetros personalizados de implantação.

  python.fn-execution.bundle.size: Default value: 100000. Unit: records.
  python.fn-execution.bundle.time: Default value: 1000. Unit: milliseconds.

  Para obter mais informações sobre esses parâmetros, consulte Flink Python Configuration.

Solução de problemas de exceções de checkpoint

1. Diagnosticar o tipo de exceção

   Visualize o histórico de checkpoints na aba Alarm ou State para identificar o tipo de exceção, como tempo limite ou falha de gravação.

   Selecione a aba Overview e expanda a seção Checkpoint. A tabela exibe detalhes de cada checkpoint, incluindo ID, status, horário de acionamento, duração e tamanho dos dados. A coluna Status mostra se cada checkpoint teve êxito ou falhou.

2. Isolar e resolver o problema

   • Cenário 1: Tempos limite frequentes de checkpoint. Verifique se há contrapressão na implantação. Analise a causa raiz da contrapressão, identifique o operador lento e resolva o problema ajustando recursos ou configurações. Para obter mais informações, consulte Como solucionar problemas de contrapressão.

   • Cenário 2: Falhas de gravação de checkpoint. Siga estas etapas para localizar os logs relevantes do TaskManager e analisar a causa raiz.

     1. Na página Checkpoints da aba Logs, clique em Checkpoints History.

        Na página Checkpoints History, visualize detalhes de cada checkpoint, como ID, Status, Acknowledged, Trigger Time, End to End Duration e Checkpointed Data Size.

     2. Clique no sinal de mais (+) ao lado do checkpoint com falha para visualizar os detalhes do operador.

     3. Expanda o operador com falha e clique em SubTask ID para acessar os logs correspondentes do TaskManager.

Erro: Restauração de estado antigo com mecanismo V4

• Mensagem de erro

  Ao atualizar do VVR 6.x para o VVR 8.x, você pode encontrar o erro: You are using the new V4 state engine to restore old state data from a checkpoint

• Causa

  O VVR 6.x e o VVR 8.x usam versões diferentes do GeminiStateBackend, e seus checkpoints não são compatíveis.

• Solução

  Use qualquer um dos métodos a seguir para resolver este problema:

  • Crie um savepoint no formato padrão e inicie a implantação a partir desse estado. Para obter mais informações, consulte Criar manualmente um savepoint e Iniciar uma implantação.

  • Reinicie a implantação sem estado.

  • (Não recomendado) Continue usando a versão legada do Gemini. Defina o parâmetro state.backend.gemini.engine.type: STREAMING e reinicie a implantação para que a alteração tenha efeito. Para obter informações sobre como configurar parâmetros, consulte Como configurar parâmetros de implantação.

  • (Não recomendado) Continue usando o mecanismo VVR 6.x para iniciar a implantação.

Erro: java.lang.NegativeArraySizeException

• Mensagem de erro

  Uma implantação que usa estado de lista pode encontrar a seguinte exceção durante a execução:

  Caused by: java.lang.NegativeArraySizeException
    at com.alibaba.gemini.engine.rm.GUnPooledByteBuffer.newTempBuffer(GUnPooledByteBuffer.java:270)
    at com.alibaba.gemini.engine.page.bmap.BinaryValue.merge(BinaryValue.java:85)
    at com.alibaba.gemini.engine.page.bmap.BinaryValue.merge(BinaryValue.java:75)
    at com.alibaba.gemini.engine.pagestore.PageStoreImpl.internalGet(PageStoreImpl.java:428)
    at com.alibaba.gemini.engine.pagestore.PageStoreImpl.get(PageStoreImpl.java:271)
    at com.alibaba.gemini.engine.pagestore.PageStoreImpl.get(PageStoreImpl.java:112)
    at com.alibaba.gemini.engine.table.BinaryKListTable.get(BinaryKListTable.java:118)
    at com.alibaba.gemini.engine.table.BinaryKListTable.get(BinaryKListTable.java:57)
    at com.alibaba.flink.statebackend.gemini.subkeyed.GeminiSubKeyedListStateImpl.getOrDefault(GeminiSubKeyedListStateImpl.java:97)
    at com.alibaba.flink.statebackend.gemini.subkeyed.GeminiSubKeyedListStateImpl.get(GeminiSubKeyedListStateImpl.java:88)
    at com.alibaba.flink.statebackend.gemini.subkeyed.GeminiSubKeyedListStateImpl.get(GeminiSubKeyedListStateImpl.java:47)
    at com.alibaba.flink.statebackend.gemini.context.ContextSubKeyedListState.get(ContextSubKeyedListState.java:60)
    at com.alibaba.flink.statebackend.gemini.context.ContextSubKeyedListState.get(ContextSubKeyedListState.java:44)
    at org.apache.flink.streaming.runtime.operators.windowing.WindowOperator.onProcessingTime(WindowOperator.java:533)
    at org.apache.flink.streaming.api.operators.InternalTimerServiceImpl.onProcessingTime(InternalTimerServiceImpl.java:289)
    at org.apache.flink.streaming.runtime.tasks.StreamTask.invokeProcessingTimeCallback(StreamTask.java:1435)

• Causa

  Os dados de estado de uma única chave no estado de lista excederam 2 GB. Isso pode ocorrer da seguinte forma:

  1. Durante a operação normal, valores anexados a uma única chave em um estado de lista são combinados por meio de um processo de mesclagem (por exemplo, em um operador de janela), fazendo com que os dados de estado cresçam continuamente.

  2. Quando os dados de estado atingem um determinado tamanho, podem primeiro acionar um erro de Out Of Memory (OOM). Após a recuperação da implantação da falha, o processo de mesclagem pode fazer com que o backend de estado solicite um array de bytes temporário que excede 2 GB, resultando nesta exceção.

  Nota

  O RocksDBStateBackend pode encontrar um problema semelhante, que pode acionar um ArrayIndexOutOfBoundsException ou uma falha de segmentação. Para obter mais informações, consulte The EmbeddedRocksDBStateBackend.

• Solução

  • Se o estado grande for causado por um operador de janela, considere reduzir o tamanho da janela.

  • Se o estado grande for causado pela lógica da implantação, considere redesenhá-la, por exemplo, dividindo as chaves.

Erro: FlinkKafkaException: Too many ongoing snapshots

• Mensagem de erro

  org.apache.flink.streaming.connectors.kafka.FlinkKafkaException: Too many ongoing snapshots. Increase kafka producers pool size or decrease number of concurrent checkpoints

• Causa

  Este erro ocorre ao usar um sink Kafka e é causado por múltiplas falhas consecutivas de checkpoint.

• Solução

  Para evitar falhas devido a tempos limite, aumente o tempo limite do checkpoint ajustando o parâmetro execution.checkpointing.timeout. Para obter mais informações sobre como configurar parâmetros, consulte Configurar parâmetros personalizados de implantação.

Erro: Exceeded checkpoint tolerable failure threshold

• Mensagem de erro

  org.apache.flink.util.FlinkRuntimeException:Exceeded checkpoint tolerable failure threshold.
    at org.apache.flink.runtime.checkpoint.CheckpointFailureManager.handleJobLevelCheckpointException(CheckpointFailureManager.java:66)

• Causa

  O número configurado de falhas de checkpoint toleráveis é muito baixo, fazendo com que a implantação acione um failover quando esse limiar é excedido. Se este parâmetro não for definido, o valor padrão é 0, o que significa que nenhuma falha de checkpoint é tolerada.

• Solução

  Ajuste o número de falhas de checkpoint permitidas definindo o parâmetro execution.checkpointing.tolerable-failed-checkpoints: num, onde num deve ser 0 ou um número inteiro positivo. Para obter mais informações sobre como configurar parâmetros, consulte Configurar parâmetros personalizados de implantação.