Utilice la biblioteca de clientes de Kinesis

Se conecta al flujo de datos.

Enumera las particiones del flujo de datos.

Utiliza los arrendamientos para coordinar las asociaciones de fragmentos con sus trabajadores

Crea instancias de un procesador de registros para cada fragmento que administra

Extrae registros de datos del flujo de datos.

Inserta los registros en el procesador de registros correspondiente

Genera puntos de comprobación para los registros procesados

Equilibra las asociaciones entre procesos de trabajo y particiones (arrendamientos) cuando cambia el recuento de instancias de procesos de trabajo o cuando se vuelve a realizar la partición del flujo de datos (las particiones se dividen o fusionan).

KCL1.x

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

KCL2.x

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

KCLaplicación para consumidores: una aplicación creada a medida que utiliza KCL y está diseñada para leer y procesar registros de flujos de datos.

Instancia de aplicación de KCL consumo: las aplicaciones de consumo 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 registros de datos.

Trabajador: clase de alto nivel que utiliza una instancia de aplicación de KCL consumo 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 trabajador KCL proporciona 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 KCL consumo y las AWS credenciales necesarias para acceder a este flujo de datos. El empleado también pone en marcha esa instancia de aplicación de KCL consumo específica 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 distribuidas para KCL consumidores utilizan los arrendamientos para dividir el procesamiento de registros de datos entre una flota de trabajadores. En un momento dado, cada fragmento de registro de datos está vinculado a un trabajador en particular mediante un contrato de arrendamiento identificado por la leaseKeyvariable.

De forma predeterminada, un trabajador puede tener uno o más contratos de arrendamiento (sujetos al valor de la variable maxLeasesForTrabajador) 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 KCL repositorios 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 la versión 1.x y /blob/master/ /src/main/java/software/amazon/kinesis/leases/Lease.java para la versión 2.x. KCL https://github.com/awslabs/ amazon-kinesis-client amazon-kinesis-client KCL

Tabla de arrendamiento: una tabla exclusiva de Amazon DynamoDB que se utiliza para realizar un seguimiento de los fragmentos de KDS un flujo de datos que los trabajadores de la aplicación de consumo están arrendando y procesando. KCL La tabla de arrendamientos debe permanecer sincronizada (dentro de un trabajador y entre todos los trabajadores) con la información más reciente sobre los fragmentos del flujo de datos mientras se ejecuta la aplicación para consumidores. KCL Para obtener más información, consulte Utilice una tabla de arrendamientos para realizar un seguimiento de los fragmentos procesados por la aplicación de consumo KCL.

Procesador de registros: la lógica que define la forma en que su aplicación de KCL consumo procesa los datos que obtiene de los flujos de datos. En tiempo de ejecución, una instancia de aplicación de KCL consumo crea una instancia de un trabajador, y este trabajador crea una instancia de un procesador de registros por cada fragmento que tenga 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.

checkpointSubSequenceNúmero: cuando se utiliza la función de agregación de la biblioteca de productores de Kinesis, se trata de una extensión del punto de control que rastrea los registros de los usuarios individuales dentro del registro de Kinesis.

leaseCounter: Se utiliza para controlar las versiones de arrendamiento, de modo que los trabajadores puedan detectar que otro trabajador ha contratado su arrendamiento.

leaseKey: un identificador único para un arrendamiento. 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 trabajador titular de este contrato de arrendamiento.

ownerSwitchesSincePunto de control: cuántas veces este contrato de arrendamiento ha cambiado de trabajador desde la última vez que se firmó un punto de control.

parentShardId: Se utiliza para garantizar que el fragmento principal se procese por completo antes de que comience 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 arrendamientos está vacía, KCL utiliza la opción de filtrado correspondiente (el parámetro ListShard API de solicitud ShardFilter opcional) para recuperar y crear arrendamientos únicamente para una instantánea de los fragmentos abiertos en el momento especificado por el parámetro. ShardFilter El ShardFilter parámetro permite filtrar la respuesta de. ListShards API La única propiedad obligatoria del parámetro ShardFilter es Type. KCLutiliza la propiedad de Type filtro y los siguientes valores válidos para identificar y devolver una instantánea de los fragmentos abiertos que podrían requerir nuevas concesiones:

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.

KCL2.3 usa el parámetro de ChildShards retorno de GetRecords y el SubscribeToShard APIs para realizar la sincronización entre arrendamiento y fragmento, como ocurre en SHARD_END el caso de los fragmentos cerrados, lo que permite a un KCL trabajador crear arrendamientos solo para los fragmentos secundarios del fragmento que ha terminado de procesar. Para compartirlo en todas las aplicaciones de consumo, esta optimización de la sincronización entre el arrendamiento y el fragmento utiliza el parámetro de. ChildShards GetRecords API Para las aplicaciones de consumo dedicadas al rendimiento (distribución mejorada), esta optimización de la sincronización entre arrendamiento y fragmento utiliza el parámetro de. ChildShards SubscribeToShard API Para obtener más información, consulte, y. GetRecordsSubscribeToShardsChildShard

Con los cambios anteriores, el comportamiento de KCL está pasando del modelo en el que todos los trabajadores aprenden sobre todos los fragmentos existentes al modelo en el que los trabajadores aprenden solo sobre los fragmentos secundarios de los fragmentos que son propiedad de cada trabajador. Por lo tanto, además de la sincronización que se produce durante el arranque de las aplicaciones de consumo y los eventos de refragmentación, KCL ahora también realiza escaneos periódicos adicionales de los fragmentos o arrendamientos para identificar cualquier posible vacío en la tabla de arrendamientos (en otras palabras, para obtener información sobre todos los fragmentos nuevos) a fin de garantizar que se procese todo el rango de hash del flujo de datos y crear arrendamientos para ellos si es necesario. PeriodicShardSyncManageres el componente responsable de realizar escaneos periódicos de los fragmentos o arrendamientos.

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

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

Ahora también se emiten nuevas CloudWatch métricas para monitorear el estado delPeriodicShardSyncManager. 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 arrendamientos está vacía, KCL utiliza la opción de filtrado correspondiente (el parámetro ListShard API de solicitud ShardFilter opcional) para recuperar y crear arrendamientos únicamente para una instantánea de los fragmentos abiertos en el momento especificado por el parámetro. ShardFilter El ShardFilter parámetro permite filtrar la respuesta de. ListShards API La única propiedad obligatoria del parámetro ShardFilter es Type. KCLutiliza la propiedad de Type filtro y los siguientes valores válidos para identificar y devolver una instantánea de los fragmentos abiertos que podrían requerir nuevas concesiones:

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.

KCL1.14 usa el parámetro de ChildShards retorno de GetRecords y el SubscribeToShard APIs para realizar la sincronización entre arrendamiento y fragmento, como ocurre en el caso de los fragmentos cerrados, lo que permite a un KCL trabajador crear arrendamientos únicamente SHARD_END para los fragmentos secundarios del fragmento que ha terminado de procesar. Para obtener más información, consulte y. GetRecordsChildShard

Con los cambios anteriores, el comportamiento de KCL está pasando del modelo en el que todos los trabajadores aprenden sobre todos los fragmentos existentes al modelo en el que los trabajadores aprenden solo sobre los fragmentos secundarios de los fragmentos que son propiedad de cada trabajador. Por lo tanto, además de la sincronización que se produce durante el arranque de las aplicaciones de consumo y los eventos de refragmentación, KCL ahora también realiza escaneos periódicos adicionales de los fragmentos o arrendamientos para identificar cualquier posible vacío en la tabla de arrendamientos (en otras palabras, para obtener información sobre todos los fragmentos nuevos) a fin de garantizar que se procese todo el rango de hash del flujo de datos y crear arrendamientos para ellos si es necesario. PeriodicShardSyncManageres el componente responsable de realizar escaneos periódicos de los fragmentos o arrendamientos.

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 la versión KCL 1.14, consulta PeriodicShardSyncManager https://github.com/awslabs/ amazon-kinesis-client /blob/v1.x/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/ .java #L987 -L999. KinesisClientLibConfiguration

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

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

KCLLa versión 1.14 ahora también admite la reducción de los 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.

MultistreamTracker interfaz

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

El streamConfigList método rellena la StreamConfiglista.

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 proporcionarlo consumerArn únicamente si utiliza la versión KCL 2.x para implementar una aplicación de consumo mejorada y desplegable.

Para obtener más informaciónStreamIdentifier, consulte https://github.com/awslabs/amazon-kinesis-client/blob/v2.5.8/ /src/main/java/software/amazon/kinesis/common/ .java #L129 amazon-kinesis-client. StreamIdentifier Para crear una, te recomendamos que crees una instancia multistream a partir de y la que esté disponible en la versión 2.5.0 y versiones posteriores. StreamIdentifier streamArn streamCreationEpoch En las KCL versiones 2.3 y 2.4, que no son compatiblesstreamArm, crea una instancia multisecuencia con este 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 CANNOT cambiará durante el tiempo de ejecución de la aplicación para consumidores. 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/ .java FormerStreamsLeasesDeletionStrategy

ConfigsBuilderes una clase que abarca toda la aplicación y que se puede utilizar para especificar todos los ajustes de configuración de la versión 2.x que se utilizarán al crear una aplicación de consumo. KCL KCL ConfigsBuilderla clase ahora es compatible con la interfaz. MultistreamTracker Puede inicializar ConfigsBuilder cualquiera de las dos con el nombre del flujo de datos desde el que desee consumir 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 inicializarlo ConfigsBuilder MultiStreamTracker si desea implementar una aplicación para KCL consumidores 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 el soporte multiflujo implementado para su aplicación de KCL consumo, cada fila de la tabla de arrendamiento de la aplicación ahora contiene el ID del fragmento y el nombre de la secuencia de los múltiples flujos de datos que procesa esta aplicación.

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