Información sobre KCL 1.x y 2.x

KCL 1.x

Para obtener más información, consulte Desarrollar consumidores de KCL 1.x

KCL 2.x

Para obtener más información, consulte Desarrollar consumidores de KCL 2.x

Aplicación para consumidores de KCL: una aplicación creada a medida con KCL y diseñada para leer y procesar registros de flujos de datos.

Instancia de aplicación de consumo: las aplicaciones de consumo de KCL suelen estar distribuidas y una o más instancias de aplicación se ejecutan simultáneamente para coordinar los fallos y equilibrar la carga de forma dinámica del procesamiento de registro de datos.

Proceso de trabajo: clase de alto nivel que utiliza una instancia de aplicación de consumo de KCL para empezar a procesar datos.

El proceso de trabajo inicializa y supervisa diversas tareas, como la sincronización de la información sobre los arrendamientos y las particiones, el seguimiento de las asignaciones de las particiones y el procesamiento de los datos de las particiones. Un proceso de trabajo proporciona a KCL la información de configuración de la aplicación de consumo, como el nombre del flujo de datos cuyos registros de datos va a procesar la aplicación de consumo de KCL y las credenciales de AWS necesarias para acceder a este flujo de datos. El proceso de trabajo también pone en marcha esa instancia específica de la aplicación de consumo de KCL para entregar los registros de datos del flujo de datos a los procesadores de registros.

Arrendamiento: datos que definen el enlace entre un proceso de trabajo y una partición. Las aplicaciones de consumo distribuidas de KCL utilizan los arrendamientos para dividir el procesamiento de registros de datos entre una flota de procesos de trabajo. En un momento dado, cada partición de registros de datos está vinculada a un proceso de trabajo en particular mediante un arrendamiento identificado por la variable leaseKey.

De forma predeterminada, un proceso de trabajo puede tener uno o más arrendamientos (sujetos al valor de la variable maxLeasesForWorker) al mismo tiempo.

Por ejemplo, si tiene una instancia de aplicación de consumo A con el proceso de trabajo A que procesa un flujo de datos con 4 particiones, el proceso de trabajo A puede retener los arrendamientos de las particiones 1, 2, 3 y 4 al mismo tiempo. Sin embargo, si tiene dos instancias de aplicaciones de consumo: A y B con el proceso de trabajo A y el proceso de trabajo B, y estas instancias procesan un flujo de datos con 4 particiones, el proceso de trabajo A y el proceso de trabajo B no pueden retener el arrendamiento de la partición 1 al mismo tiempo. Un proceso de trabajo retiene el arrendamiento de una partición concreta hasta que esté listo para dejar de procesar los registros de datos de esta partición o hasta que falle. Cuando un proceso de trabajo deja de ser titular del arrendamiento, otro proceso de trabajo lo acepta y lo retiene.

Para obtener más información (estos son los repositorios de KCL de Java), consulte https://github.com/awslabs/amazon-kinesis-client/blob/v1.x/src/main/java/com/amazonaws/services/kinesis/leases/impl/Lease.java para KCL 1.x y https://github.com/awslabs/amazon-kinesis-client/blob/master/amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/Lease.java para KCL 2.x.

Tabla de arrendamiento: una tabla exclusiva de Amazon DynamoDB que se utiliza para realizar un seguimiento de las particiones de un flujo de datos de KDS que los procesos de trabajo de la aplicación de consumo de KCL están arrendando y procesando. La tabla de arrendamiento debe permanecer sincronizada (dentro de un proceso de trabajo y entre todos los procesos de trabajo) con la información más reciente sobre las particiones del flujo de datos mientras se ejecuta la aplicación de consumo de KCL. Para obtener más información, consulte Usar una tabla de arrendamiento para realizar el seguimiento de las particiones procesadas por la aplicación de consumo de KCL.

Procesador de registros: lógica que define la forma en que su aplicación de consumo de KCL procesa los datos que obtiene de los flujos de datos. En tiempo de ejecución, una instancia de una aplicación de consumo de KCL crea una instancia de un proceso de trabajo, y este proceso de trabajo crea una instancia de un procesador de registros por cada partición que tiene en arrendamiento.

checkpoint: el número secuencial de punto de comprobación más reciente del fragmento. Este valor es único en todas las particiones del flujo de datos.

checkpointSubSequenceNumber: cuando se utiliza la característica de agregación de Kinesis Client Library, esto es una extensión del valor checkpoint que realiza un seguimiento de los registros de los usuarios individuales en el registro de Kinesis.

leaseCounter: se utiliza para el control de versiones de las asignaciones, de modo que los procesos de trabajo puedan detectar que su asignación ha sido utilizada por otro proceso de trabajo.

leaseKey: un identificador único para una asignación. Cada arrendamiento es específico de una partición del flujo de datos y solo lo retiene un proceso de trabajo cada vez.

leaseOwner: el proceso de trabajo que tiene esta asignación.

ownerSwitchesSinceCheckpoint: veces que la asignación ha cambiado de proceso de trabajo desde la última vez que se escribió un punto de comprobación.

parentShardId: se utiliza para garantizar que el fragmento principal ha sido procesado totalmente antes de comenzar el procesamiento en los fragmentos secundarios. Así, se garantiza que los registros se procesen en el mismo orden en el que se introdujeron en la secuencia.

hashrange: lo utiliza PeriodicShardSyncManager para ejecutar sincronizaciones periódicas para encontrar las particiones que faltan en la tabla de arrendamiento y crear arrendamientos para ellas si es necesario.

childshards: lo utiliza LeaseCleanupManager para revisar el estado de procesamiento de la partición secundaria y decidir si la partición principal se puede eliminar de la tabla de arrendamiento.

shardID: ID de la partición.

nombre del flujo El identificador del flujo de datos en el siguiente formato: account-id:StreamName:streamCreationTimestamp.

Obtiene todos las particiones de datos del flujo que se está procesando.

Obtiene todos los arrendamientos de particiones de la tabla de arrendamiento.

Filtra cada partición abierta que no tenga arrendamientos en la tabla de arrendamiento.

Repite todas las particiones abiertas encontradas y para cada partición abierta sin un elemento principal abierto:

Durante el arranque de la aplicación, si la tabla de arrendamiento está vacía, KCL utiliza la opción de filtrado de la API ListShard (el parámetro de solicitud ShardFilter opcional) para recuperar y crear arrendamientos únicamente para una instantánea de las particiones abiertas en el momento especificado por el parámetro ShardFilter. El parámetro ShardFilter permite filtrar la respuesta de la API ListShards. La única propiedad obligatoria del parámetro ShardFilter es Type. KCL utiliza la propiedad de filtro Type y los siguientes valores válidos para identificar y devolver una instantánea de las particiones abiertas que podrían requerir nuevos arrendamientos:

ShardFilter se utiliza al crear arrendamientos para una tabla de arrendamiento vacía con el fin de inicializar los arrendamientos de una instantánea de las particiones especificadas en RetrievalConfig#initialPositionInStreamExtended.

Para obtener más información acerca de ShardFilter, consulte https://docs.aws.amazon.com/kinesis/latest/APIReference/API_ShardFilter.html.

En lugar de que todos los procesos de trabajo realicen la sincronización entre arrendamientos y particiones para mantener la tabla de arrendamiento actualizada con las particiones más recientes del flujo de datos, un único líder del proceso de trabajo elegido realiza la sincronización entre arrendamientos y particiones.

KCL 2.3 utiliza el parámetro de devolución ChildShards de las API GetRecords y SubscribeToShard para realizar la sincronización entre arrendamientos y particiones que se produce en SHARD_END para las particiones cerradas, lo que permite a un proceso de trabajo de KCL crear arrendamientos solo para las particiones secundarias de la partición que ha terminado de procesar. Para compartirla entre aplicaciones de consumo, esta optimización de la sincronización entre arrendamientos y particiones utiliza el parámetro ChildShards de la API GetRecords. En el caso de las aplicaciones de consumo dedicadas al rendimiento (distribución mejorada), esta optimización de la sincronización entre arrendamientos y particiones utiliza el parámetro ChildShards de la API SubscribeToShard. Para obtener más información, consulte GetRecords, SubscribeToShards y ChildShard.

Con los cambios anteriores, el comportamiento de KCL está pasando del modelo en el que todos los procesos de trabajo aprenden sobre todas las particiones existentes al modelo en el que los procesos de trabajo aprenden solo sobre las particiones secundarias de las particiones que son propiedad de cada proceso de trabajo. Por lo tanto, además de la sincronización que se produce durante el arranque de las aplicaciones de consumo y los eventos de repartición, KCL ahora también realiza exámenes periódicos adicionales de las particiones o arrendamientos para identificar cualquier posible laguna en la tabla de arrendamiento (en otras palabras, para obtener información sobre todas las particiones nuevas) a fin de garantizar que se procese todo el rango de hash del flujo de datos y crear arrendamientos para ellos si es necesario. PeriodicShardSyncManager es el componente responsable de ejecutar exámenes periódicos de arrendamientos o particiones.

Para obtener más información sobre PeriodicShardSyncManager en KCL 2.3, consulte https://github.com/awslabs/amazon-kinesis-client/blob/master/amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/LeaseManagementConfig.java#L201-L213.

En KCL 2.3, hay nuevas opciones de configuración disponibles para configurar PeriodicShardSyncManager en LeaseManagementConfig:

Ahora también se emiten nuevas métricas de CloudWatch para supervisar el estado de PeriodicShardSyncManager. Para obtener más información, consulte PeriodicShardSyncManager.

Incluye una optimización de HierarchicalShardSyncer para crear solo arrendamientos para una capa de particiones.

Durante el arranque de la aplicación, si la tabla de arrendamiento está vacía, KCL utiliza la opción de filtrado de la API ListShard (el parámetro de solicitud ShardFilter opcional) para recuperar y crear arrendamientos únicamente para una instantánea de las particiones abiertas en el momento especificado por el parámetro ShardFilter. El parámetro ShardFilter permite filtrar la respuesta de la API ListShards. La única propiedad obligatoria del parámetro ShardFilter es Type. KCL utiliza la propiedad de filtro Type y los siguientes valores válidos para identificar y devolver una instantánea de las particiones abiertas que podrían requerir nuevos arrendamientos:

ShardFilter se utiliza al crear arrendamientos para una tabla de arrendamiento vacía con el fin de inicializar los arrendamientos de una instantánea de las particiones especificadas en KinesisClientLibConfiguration#initialPositionInStreamExtended.

Para obtener más información acerca de ShardFilter, consulte https://docs.aws.amazon.com/kinesis/latest/APIReference/API_ShardFilter.html.

En lugar de que todos los procesos de trabajo realicen la sincronización entre arrendamientos y particiones para mantener la tabla de arrendamiento actualizada con las particiones más recientes del flujo de datos, un único líder del proceso de trabajo elegido realiza la sincronización entre arrendamientos y particiones.

KCL 1.14 utiliza el parámetro de devolución ChildShards de las API GetRecords y SubscribeToShard para realizar la sincronización entre arrendamientos y particiones que se produce en SHARD_END para las particiones cerradas, lo que permite a un proceso de trabajo de KCL crear arrendamientos solo para las particiones secundarias de la partición que ha terminado de procesar. Para obtener más información, consulte GetRecords y ChildShard.

Con los cambios anteriores, el comportamiento de KCL está pasando del modelo en el que todos los procesos de trabajo aprenden sobre todas las particiones existentes al modelo en el que los procesos de trabajo aprenden solo sobre las particiones secundarias de las particiones que son propiedad de cada proceso de trabajo. Por lo tanto, además de la sincronización que se produce durante el arranque de las aplicaciones de consumo y los eventos de repartición, KCL ahora también realiza exámenes periódicos adicionales de las particiones o arrendamientos para identificar cualquier posible laguna en la tabla de arrendamiento (en otras palabras, para obtener información sobre todas las particiones nuevas) a fin de garantizar que se procese todo el rango de hash del flujo de datos y crear arrendamientos para ellos si es necesario. PeriodicShardSyncManager es el componente responsable de ejecutar exámenes periódicos de arrendamientos o particiones.

Cuando KinesisClientLibConfiguration#shardSyncStrategyType está establecido en ShardSyncStrategyType.SHARD_END, PeriodicShardSync leasesRecoveryAuditorInconsistencyConfidenceThreshold se utiliza para determinar el umbral del número de exámenes consecutivos que contienen lagunas en la tabla de arrendamiento, tras lo cual se exige la sincronización de las particiones. Cuando KinesisClientLibConfiguration#shardSyncStrategyType se establece en ShardSyncStrategyType.PERIODIC, leasesRecoveryAuditorInconsistencyConfidenceThreshold se ignora.

Para obtener más información sobre PeriodicShardSyncManager en KCL 1.14, consulte https://github.com/awslabs/amazon-kinesis-client/blob/v1.x/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/KinesisClientLibConfiguration.java#L987-L999.

En KCL 1.14, hay una nueva opción de configuración disponible para configurar PeriodicShardSyncManager en LeaseManagementConfig:

Ahora también se emiten nuevas métricas de CloudWatch para supervisar el estado de PeriodicShardSyncManager. Para obtener más información, consulte PeriodicShardSyncManager.

KCL 1.14 ahora también admite la limpieza de arrendamientos diferidos. LeaseCleanupManager elimina los arrendamientos de forma asíncrona al llegar a SHARD_END, cuando una partición ha caducado pasado el periodo de retención del flujo de datos o se ha cerrado como resultado de una operación de repartición.

Están disponibles nuevas opciones de configuración para LeaseCleanupManager.

Incluye una optimización de KinesisShardSyncer para crear solo arrendamientos para una capa de particiones.

Interfaz MultistreamTracker

Para crear una aplicación de consumo que pueda procesar varios flujos al mismo tiempo, debe implementar una nueva interfaz llamada MultistreamTracker. Esta interfaz incluye el método streamConfigList que devuelve la lista de flujos de datos y sus configuraciones para que los procese la aplicación de consumo de KCL. Tenga en cuenta que los flujos de datos que se procesan pueden cambiar durante el tiempo de ejecución de la aplicación de consumo. KCL llama a streamConfigList periódicamente para obtener información sobre los cambios en los flujos de datos que se van a procesar.

El método streamConfigList rellena la lista StreamConfig.

package software.amazon.kinesis.common;

import lombok.Data;
import lombok.experimental.Accessors;

@Data
@Accessors(fluent = true)
public class StreamConfig {
    private final StreamIdentifier streamIdentifier;
    private final InitialPositionInStreamExtended initialPositionInStreamExtended;
    private String consumerArn;
}
                

Tenga en cuenta que los campos StreamIdentifier y InitialPositionInStreamExtended son obligatorios, aunque consumerArn es opcional. Debe proporcionar consumerArn únicamente si utiliza KCL 2.x para implementar una aplicación de consumo con distribución mejorada.

Para obtener más información sobre StreamIdentifier, consulte https://github.com/awslabs/amazon-kinesis-client/blob/v2.5.8/amazon-kinesis-client/src/main/java/software/amazon/kinesis/common/StreamIdentifier.java#L129. Para crear un StreamIdentifier, le recomendamos que cree una instancia multiflujo a partir de streamArn y la streamCreationEpoch que esté disponible en la versión 2.5.0 y versiones posteriores. En las versiones KCL 2.3 y 2.4, que no son compatibles con streamArm, cree una instancia multiflujo con el formato account-id:StreamName:streamCreationTimestamp. Este formato quedará obsoleto y dejará de ser compatible a partir de la próxima versión principal.

MultistreamTracker también incluye una estrategia para eliminar los arrendamientos de flujos antiguos en la tabla de arrendamiento (formerStreamsLeasesDeletionStrategy). Tenga en cuenta que la estrategia NO SE PUEDE cambiar durante el tiempo de ejecución de la aplicación de consumo. Para obtener más información, consulte https://github.com/awslabs/amazon-kinesis-client/blob/0c5042dadf794fe988438436252a5a8fe70b6b0b/amazon-kinesis-client/src/main/java/software/amazon/kinesis/processor/FormerStreamsLeasesDeletionStrategy.java.

ConfigsBuilder es una clase que abarca toda la aplicación y que puede utilizar para especificar todos los ajustes de configuración de KCL 2.x que se utilizarán al crear su aplicación de consumo de KCL. La clase ConfigsBuilder ahora es compatible con la interfaz de MultistreamTracker. Puede inicializar ConfigsBuilder con el nombre del flujo de datos del que desee consumir los registros:

 /**
     * Constructor to initialize ConfigsBuilder with StreamName
     * @param streamName
     * @param applicationName
     * @param kinesisClient
     * @param dynamoDBClient
     * @param cloudWatchClient
     * @param workerIdentifier
     * @param shardRecordProcessorFactory
     */
    public ConfigsBuilder(@NonNull String streamName, @NonNull String applicationName,
            @NonNull KinesisAsyncClient kinesisClient, @NonNull DynamoDbAsyncClient dynamoDBClient,
            @NonNull CloudWatchAsyncClient cloudWatchClient, @NonNull String workerIdentifier,
            @NonNull ShardRecordProcessorFactory shardRecordProcessorFactory) {
        this.appStreamTracker = Either.right(streamName);
        this.applicationName = applicationName;
        this.kinesisClient = kinesisClient;
        this.dynamoDBClient = dynamoDBClient;
        this.cloudWatchClient = cloudWatchClient;
        this.workerIdentifier = workerIdentifier;
        this.shardRecordProcessorFactory = shardRecordProcessorFactory;
    }

O bien, puede inicializar ConfigsBuilder con MultiStreamTracker si desea implementar una aplicación de consumo de KCL que procese varios flujos al mismo tiempo.

* Constructor to initialize ConfigsBuilder with MultiStreamTracker
     * @param multiStreamTracker
     * @param applicationName
     * @param kinesisClient
     * @param dynamoDBClient
     * @param cloudWatchClient
     * @param workerIdentifier
     * @param shardRecordProcessorFactory
     */
    public ConfigsBuilder(@NonNull MultiStreamTracker multiStreamTracker, @NonNull String applicationName,
            @NonNull KinesisAsyncClient kinesisClient, @NonNull DynamoDbAsyncClient dynamoDBClient,
            @NonNull CloudWatchAsyncClient cloudWatchClient, @NonNull String workerIdentifier,
            @NonNull ShardRecordProcessorFactory shardRecordProcessorFactory) {
        this.appStreamTracker = Either.left(multiStreamTracker);
        this.applicationName = applicationName;
        this.kinesisClient = kinesisClient;
        this.dynamoDBClient = dynamoDBClient;
        this.cloudWatchClient = cloudWatchClient;
        this.workerIdentifier = workerIdentifier;
        this.shardRecordProcessorFactory = shardRecordProcessorFactory;
    }                    
                

Con la compatibilidad con varios flujos implementada para la aplicación de consumo de KCL, cada fila de la tabla de arrendamiento de la aplicación ahora contiene el ID de la partición y el nombre del flujo de los varios flujos de datos que procesa esta aplicación.

Cuando se implementa la compatibilidad con varios flujos para la aplicación de consumo de KCL, leaseKey adopta la siguiente estructura: account-id:StreamName:streamCreationTimestamp:ShardId. Por ejemplo, 111111111:multiStreamTest-1:12345:shardId-000000000336.