Realtime Compute for Apache Flink:Checkpoints y savepoints

• Why is data not updated after the state TTL expires when mini-batch is enabled?

• How to calculate the start time of the next periodic checkpoint

• What are the differences between the GeminiStateBackend in VVR 8.x and VVR 6.x?

• Is it normal for a full checkpoint and an incremental checkpoint to have the same size?

• What to do if checkpoints are slow for a Python deployment

• How to troubleshoot checkpoint exceptions

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

• Error: java.lang.NegativeArraySizeException

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

• Error: Exceeded checkpoint tolerable failure threshold

Errores de actualización de datos por mini-batch

El estado conserva los resultados de cálculos completos anteriores. Si el TTL del estado expira, este se borra y dichos resultados acumulados se pierden. En consecuencia, no es posible actualizar los nuevos datos con base en los resultados del mini-batch.

Por otro lado, si el mini-batch está deshabilitado, cuando el TTL del estado expira, los datos de la clave vencida se recalculan y se emiten. Esto garantiza la actualización continua de los datos. No obstante, la mayor frecuencia de actualizaciones puede provocar otros problemas, como retrasos en el procesamiento.

Por lo tanto, configure los ajustes de mini-batch y TTL según sus requisitos específicos de negocio.

Cálculo de la hora de inicio del siguiente checkpoint

La hora de inicio del siguiente checkpoint depende de dos parámetros: el intervalo entre checkpoints y la pausa mínima entre checkpoints. Un nuevo checkpoint se activa cuando se cumplen las dos condiciones siguientes:

• Intervalo entre checkpoints: tiempo mínimo entre el inicio de un checkpoint y el inicio del siguiente. Corresponde al lapso entre <start time of previous checkpoint, start time of next checkpoint>.

• Pausa mínima: tiempo mínimo entre la finalización de un checkpoint y el inicio del siguiente. Corresponde al lapso entre <end time of previous checkpoint, start time of next checkpoint>.

Considere los dos escenarios siguientes, donde el intervalo entre checkpoints es de 3 minutos, la pausa mínima es de 3 minutos y el tiempo de espera es de 10 minutos.

• Escenario 1: El despliegue se ejecuta con normalidad y todos los checkpoints se completan correctamente.

  El primer checkpoint comienza a las 12:00:00 y finaliza correctamente a las 12:00:02. El segundo checkpoint iniciará a las 12:03:00.

• Escenario 2: El despliegue presenta un comportamiento anómalo y un checkpoint falla debido a un tiempo de espera agotado.

  El primer checkpoint comienza a las 12:00:00 y finaliza correctamente a las 12:00:02. El segundo checkpoint inicia a las 12:03:00, pero falla a las 12:13:00 por tiempo de espera agotado. El tercer checkpoint comenzará a las 12:16:00.

Para obtener más información sobre cómo configurar la pausa mínima entre checkpoints, consulte Tuning Checkpointing.

GeminiStateBackend en VVR 8.x frente a VVR 6.x

De forma predeterminada, los motores de Realtime Compute for Apache Flink para VVR 6.x utilizan GeminiStateBackend V3, mientras que los motores VVR 8.x emplean la versión V4.

Checkpoints completos e incrementales del mismo tamaño

Si observa que un checkpoint completo y uno incremental tienen el mismo tamaño, debe realizar lo siguiente:

• Compruebe que la creación de checkpoints incrementales esté configurada y habilitada correctamente.

• Este comportamiento puede ser esperado en ciertos escenarios. Por ejemplo:

  1. Antes de ingerir datos (por ejemplo, antes de las 18:29), el despliegue no ha procesado ningún dato. El checkpoint contiene únicamente el estado inicial del origen, por lo que funciona efectivamente como un checkpoint completo.

  2. A las 18:29 se ingieren un millón de registros. Si estos datos se procesan por completo dentro del siguiente intervalo de checkpoint (por ejemplo, 3 minutos) y no llegan más datos, el primer checkpoint incremental contendrá todo el estado generado a partir de esos registros.

  En este caso, se espera que el checkpoint completo y el primer checkpoint incremental tengan el mismo tamaño. El primer checkpoint incremental debe incluir el estado de todos los datos para garantizar que sea posible una recuperación completa desde ese punto, lo que lo hace funcionalmente equivalente a un checkpoint completo.

  Los beneficios de los checkpoints incrementales suelen hacerse evidentes a partir del segundo checkpoint. Con una entrada de datos estable y sin cambios importantes en el estado, los checkpoints incrementales subsiguientes deberían ser más pequeños, lo que indica que el sistema captura instantáneas solo de los cambios de estado correctamente. Si mantienen el mismo tamaño, investigue el estado y el comportamiento de su sistema para identificar posibles problemas.

Checkpoints lentos en despliegues de Python

• Causa

  Una función definida por el usuario (UDF) de Python con bajo rendimiento puede aumentar la duración del checkpoint y degradar el rendimiento del despliegue.

• Solución

  Reduzca el tamaño del búfer. En la sección Other Configuration, establezca los siguientes parámetros. Para consultar las instrucciones, vea Configure custom deployment parameters.

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

  Para obtener más información sobre estos parámetros, consulte Flink Python Configuration.

Solución de problemas de excepciones en checkpoints

1. Diagnóstico del tipo de excepción

   Consulte el historial de checkpoints en la pestaña Alarm o State para identificar el tipo de excepción, como un tiempo de espera agotado o un error de escritura.

   Seleccione la pestaña Overview y expanda la sección Checkpoint. La tabla muestra detalles de cada checkpoint, incluidos su ID, estado, hora de activación, duración y tamaño de datos. La columna Status indica si cada checkpoint se completó correctamente o falló.

2. Aislamiento y resolución del problema

   • Escenario 1: Tiempos de espera de checkpoint frecuentes. Compruebe si existe contrapresión en el despliegue. Analice la causa raíz de la contrapresión, identifique el operador lento y resuelva el problema ajustando recursos o configuraciones. Para obtener más información, consulte How to troubleshoot backpressure issues.

   • Escenario 2: Errores de escritura en checkpoints. Siga estos pasos para encontrar los registros relevantes de TaskManager y analizar la causa raíz.

     1. En la página Checkpoints de la pestaña Logs, haga clic en Checkpoints History.

        En la página Checkpoints History, puede consultar detalles de cada checkpoint, como ID, Status, Acknowledged, Trigger Time, End to End Duration y Checkpointed Data Size.

     2. Haga clic en el signo más (+) junto al checkpoint fallido para ver los detalles de su operador.

     3. Expanda el operador fallido y haga clic en el SubTask ID para ir a los registros correspondientes de TaskManager.

Error: Restauración de estado antiguo con el motor V4

• Mensaje de error

  Al actualizar de VVR 6.x a VVR 8.x, puede encontrar el error: You are using the new V4 state engine to restore old state data from a checkpoint

• Causa

  VVR 6.x y VVR 8.x utilizan versiones diferentes de GeminiStateBackend, y sus checkpoints no son compatibles.

• Solución

  Puede utilizar cualquiera de los siguientes métodos para resolver este problema:

  • Cree un savepoint en formato estándar e inicie el despliegue desde este estado. Para obtener más información, consulte Manually create a savepoint y Start a deployment.

  • Reinicie el despliegue sin estado.

  • (No recomendado) Continúe usando la versión heredada de Gemini. Debe establecer el parámetro state.backend.gemini.engine.type: STREAMING y reiniciar el despliegue para que el cambio surta efecto. Para obtener información sobre cómo configurar parámetros, consulte How to configure deployment parameters.

  • (No recomendado) Continúe usando el motor VVR 6.x para iniciar el despliegue.

Error: java.lang.NegativeArraySizeException

• Mensaje de error

  Un despliegue que utiliza un estado de lista puede encontrar la siguiente excepción durante la ejecución:

  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

  Los datos de estado de una sola clave en el estado de lista han superado los 2 GB. Esto puede ocurrir de la siguiente manera:

  1. Durante la operación normal, los valores anexados a una sola clave en un estado de lista se combinan mediante un proceso de fusión (por ejemplo, en un operador de ventana), lo que provoca que los datos de estado crezcan continuamente.

  2. Cuando los datos de estado alcanzan cierto tamaño, pueden desencadenar primero un error de Out Of Memory (OOM). Después de que el despliegue se recupera del fallo, el proceso de fusión puede hacer que el backend de estado solicite una matriz de bytes temporal que supere los 2 GB, lo que genera esta excepción.

  Nota

  RocksDBStateBackend puede experimentar un problema similar, que podría activar una excepción ArrayIndexOutOfBoundsException o un fallo de segmentación. Para obtener más información, consulte The EmbeddedRocksDBStateBackend.

• Solución

  • Si el estado grande se debe a un operador de ventana, considere reducir el tamaño de la ventana.

  • Si el estado grande se debe a la lógica del despliegue, considere rediseñarla; por ejemplo, dividiendo las claves.

Error: FlinkKafkaException: Too many ongoing snapshots

• Mensaje de error

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

• Causa

  Este error ocurre al usar un sink de Kafka y se debe a múltiples fallos consecutivos de checkpoints.

• Solución

  Para evitar fallos por tiempos de espera agotados, aumente el tiempo de espera del checkpoint ajustando el parámetro execution.checkpointing.timeout. Para obtener más información sobre cómo configurar parámetros, consulte Configure custom deployment parameters.

Error: Exceeded checkpoint tolerable failure threshold

• Mensaje de error

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

• Causa

  El número configurado de fallos de checkpoint tolerables es demasiado bajo, lo que provoca que el despliegue active un failover cuando se supera este umbral. Si este parámetro no se establece, el valor predeterminado es 0, lo que significa que no se tolera ningún fallo de checkpoint.

• Solución

  Ajuste el número de fallos de checkpoint permitidos estableciendo el parámetro execution.checkpointing.tolerable-failed-checkpoints: num, donde num debe ser 0 o un entero positivo. Para obtener más información sobre cómo configurar parámetros, consulte Configure custom deployment parameters.