Administración de registros duplicados

Hay dos principales motivos por los que se podrían entregar registros más de una vez en la aplicación de Amazon Kinesis Data Streams: los reintentos de los productores y de los consumidores. Su aplicación debe prever y administrar de forma adecuada el procesamiento de registros individuales varias veces.

Reintentos en los productores

Piense en un productor que experimenta un tiempo de inactividad relacionado con un problema en la red después de hacer una llamada a PutRecord, pero antes de que pueda recibir la confirmación desde Amazon Kinesis Data Streams. El productor no puede asegurar que se ha entregado el registro a Kinesis Data Streams. Suponiendo que cada registro es importante para la aplicación, el productor habría sido escrito para volver a intentar la llamada con los mismos datos. Si ambas llamadas a PutRecord sobre los mismos datos se enviaron correctamente a Kinesis Data Streams, habrá dos registros de Kinesis Data Streams. Aunque los dos registros tendrán datos idénticos, también tendrán números secuenciales únicos. Las aplicaciones que necesitan garantías estrictas deben tener incrustada una clave maestra en el registro para eliminar los duplicados en etapas posteriores del procesamiento. Tenga en cuenta que el número de duplicados debidos a reintentos del productor suele ser bajo en comparación con el número de duplicados debidos a reintentos del consumidor.

nota

Si utilizas el AWS SDKPutRecord, obtén información sobre el comportamiento de los reintentos del SDK en la guía del usuario de los AWS SDK y las herramientas.

Reintentos en los consumidores

Los reintentos en los consumidores (aplicaciones de procesamiento de datos) ocurren cuando se reinician los procesadores de registros. Los procesadores de registros para un mismo fragmento se reinician en los siguientes casos:

1. Si un proceso de trabajo termina de forma inesperada

2. Si las instancias de los procesos de trabajo se agregan o eliminan

3. Si se fusionan o dividen fragmentos

4. Si se implementa la aplicación

En todos estos casos, el mapeo del shards-to-worker-to procesador de registros se actualiza continuamente para equilibrar la carga del procesamiento. Los procesadores de fragmentos que se migraron a otras instancias reinician el procesamiento de registros a partir del último punto de comprobación. Esto se traduce en un procesamiento de registros duplicado, tal y como se muestra en el ejemplo siguiente. Para obtener más información sobre el balanceo de carga, consulte Cambio en los fragmentos, escalado y procesamiento paralelo.

Ejemplo: Reintentos de consumidores que dan como resultado la entrega doble de registros

En este ejemplo tiene una aplicación que lee continuamente los registros de un flujo, agrega registros en un archivo local y carga el archivo en Amazon S3. Para simplificar, supongamos que solo hay 1 fragmento y un proceso de trabajo que lo procesa. Fíjese en el siguiente ejemplo de secuencia de eventos, suponiendo que el último punto de comprobación se realizó en el número de registro 10 000:

1. Un proceso de trabajo lee el siguiente lote de registros del fragmento, los registros de 10 001 a 20 000.

2. A continuación, el proceso de trabajo transmite el lote de registros al procesador de registros asociado.

3. El procesador de registros agrega los datos, crea un archivo de Amazon S3 y carga el archivo a Amazon S3 correctamente.

4. El proceso de trabajo se cierra de forma inesperada antes de que pueda crearse un nuevo punto de comprobación.

5. La aplicación, el proceso de trabajo y el procesador de registros se reinician.

6. A partir de ahora, el proceso de trabajo comienza a leer desde el último punto de comprobación correcto, en este caso el 10 001.

Por lo tanto, los registros 10 001-20 000 se consumen más de una vez.

Resistencia a los reintentos en los consumidores

Aunque puede que los registros se procesen más de una vez, la aplicación podría presentar los efectos adversos como si los registros se hubieran procesado solo una vez (procesamiento idempotente). Las soluciones a este problema varían en cuanto a su complejidad y precisión. Si el destino final de los datos puede administrar bien los duplicados, le recomendamos que deje que dicho destino final se encargue del procesamiento idempotente. Por ejemplo, con Opensearch puede utilizar una combinación de control de versiones e ID únicos para evitar el procesamiento duplicado.

En el ejemplo de la aplicación de la sección anterior, lee continuamente los registros de un flujo, agrega los registros en un archivo local y carga el archivo en Amazon S3. Como se ha indicado, los registros 10 001-20 000 se consumen más de una vez, lo que da como resultado que haya varios archivos en Amazon S3 con los mismos datos. Una forma de mitigar los duplicados en este ejemplo consiste en garantizar que en el paso 3 se utilice el siguiente esquema:

1. El procesador de registros utiliza un número fijo de registros por archivo de Amazon S3; por ejemplo, 5000.

2. El nombre de archivo utiliza este esquema: prefijo de Amazon S3, ID de partición y First-Sequence-Num. En este caso, podría ser algo parecido a sample-shard000001-10001.

3. Después de cargar el archivo de Amazon S3, cree un punto de verificación especificando Last-Sequence-Num. En este caso, debería establecer puntos de comprobación en el número de registro 15 000.

Con este esquema, incluso si los registros se procesan más de una vez, el archivo de Amazon S3 resultante tendrá el mismo nombre y los mismos datos. Los reintentos solo dan como resultado la escritura de los mismos datos en el mismo archivo más de una vez.

En el caso de una operación de cambio en los fragmentos, el número de registros que quedan en el fragmento podría ser menor que el número fijo necesario. En este caso, el método shutdown() tiene que volcar el archivo a Amazon S3 y establecer puntos de verificación en el último número secuencial. El esquema anterior también es compatible con las operaciones de cambios en los fragmentos.