KCLInformations 1.x et 2.x

KCL1. x

Pour plus d’informations, consultez Développez des KCL consommateurs 1.x.

KCL2. x

Pour plus d’informations, consultez Développez des KCL consommateurs 2.x.

KCLapplication grand public : application conçue sur mesure pour lire KCL et traiter des enregistrements à partir de flux de données.

Instance d'application KCL grand public : les applications grand public sont généralement distribuées, une ou plusieurs instances d'application s'exécutant simultanément afin de coordonner les défaillances et d'équilibrer dynamiquement la charge du traitement des enregistrements de données.

Travailleur : classe de haut niveau utilisée par une instance d'application KCL grand public pour commencer à traiter des données.

L'application de travail initialise et supervise diverses tâches, y compris la synchronisation des informations de partition et de bail, le suivi des affectations de partitions et le traitement des données provenant des partitions. Un travailleur fournit KCL les informations de configuration de l'application client, telles que le nom du flux de données dont cette application KCL client va traiter les enregistrements de données et les AWS informations d'identification nécessaires pour accéder à ce flux de données. Le travailleur lance également cette instance d'application KCL client spécifique pour fournir des enregistrements de données du flux de données aux processeurs d'enregistrements.

Bail : données qui définissent le lien entre une application de travail et une partition. Les applications KCL grand public distribuées utilisent des contrats de location pour répartir le traitement des enregistrements de données entre un parc de travailleurs. À tout moment, chaque fragment d'enregistrements de données est lié à un travailleur en particulier par un bail identifié par la leaseKeyvariable.

Par défaut, un travailleur peut détenir un ou plusieurs baux (sous réserve de la valeur de la variable maxLeasesForWorker) en même temps.

Par exemple, si vous avez une instance d'application consommateur A avec l'application de travail A qui traite un flux de données contenant 4 partitions, l'application de travail A peut posséder simultanément des baux pour les partitions 1, 2, 3 et 4. Cependant, si vous avez deux instances d'applications consommateur, nommées A et B, chacune avec sa propre application de travail (application de travail A et application de travail B), traitant un flux de données composé de 4 partitions, alors l'application de travail A et l'application de travail B ne peuvent pas détenir simultanément le bail de la partition 1. Une application de travail possède le bail d'une partition spécifique jusqu'à ce qu'elle soit prête à arrêter de traiter les enregistrements de données de cette partition ou jusqu'à ce qu'elle rencontre une défaillance. Lorsqu'une application de travail cesse de possède le bail, une autre application de travail prend et possède ce bail.

Pour plus d'informations (il s'agit des KCL référentiels Java), consultez https://github.com/awslabs/amazon-kinesis-client/blob/v1.x/src/main/java/com/amazonaws/services/kinesis/leases/impl/Lease.java pour KCL 1.x et https://github.com/awslabs/amazon-kinesis-client/blob/master/amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/Lease.java pour 2.x. KCL

Table des baux : table Amazon DynamoDB unique utilisée pour suivre les partitions d'KDSun flux de données louées et traitées par les employés de l'application client. KCL La table des baux doit rester synchronisée (au sein d'un travailleur et entre tous les travailleurs) avec les dernières informations relatives aux partitions provenant du flux de données pendant l'exécution de l'application KCL client. Pour de plus amples informations, veuillez consulter Utilisez un tableau des baux pour suivre les partitions traitées par l'application KCL client.

Processeur d'enregistrement : logique qui définit la manière dont votre application KCL client traite les données issues des flux de données. Au moment de l'exécution, une instance d'application KCL grand public instancie un travailleur, qui instancie un processeur d'enregistrement pour chaque partition qu'il loue.

checkpoint : le plus récent numéro de séquence de point de contrôle de la partition. Cette valeur est unique dans toutes les partitions du flux de données.

checkpointSubSequenceNuméro : lorsque vous utilisez la fonctionnalité d'agrégation de la bibliothèque Kinesis Producer, il s'agit d'une extension du point de contrôle qui permet de suivre les enregistrements utilisateur individuels au sein de l'enregistrement Kinesis.

leaseCounter: utilisé pour le contrôle des versions de bail afin que les travailleurs puissent détecter que leur bail a été pris par un autre travailleur.

leaseKey: identifiant unique pour un bail. Chaque bail est propre à une partition du flux de données et est détenu par une seule application de travail à la fois.

leaseOwner: Le travailleur titulaire de ce bail.

ownerSwitchesSincePoint de contrôle : combien de fois ce bail a changé de travailleur depuis la dernière fois qu'un point de contrôle a été écrit.

parentShardId: Utilisé pour garantir que la partition parent est entièrement traitée avant le début du traitement sur les partitions enfants. Cela garantit que les enregistrements sont traités dans l'ordre dans lequel ils ont été placés dans le flux.

hashrange : utilisé par le PeriodicShardSyncManager pour exécuter des synchronisations périodiques afin de trouver les partitions manquantes dans la table des baux et de créer des baux pour celles-ci si nécessaire.

childshards : utilisé par le LeaseCleanupManager pour vérifier l'état de traitement de la partition enfant et décider si la partition parent peut être supprimée de la table des baux.

shardID : ID de la partition.

nom du flux : identifiant du flux de données au format suivant : account-id:StreamName:streamCreationTimestamp.

Récupère toutes les partitions contenant les données du flux en cours de traitement

Récupère tous les baux de partitions à partir de la table des baux

Filtre chaque partition ouverte qui n'a pas de bail dans la table des baux

Effectue une itération sur toutes les partitions ouvertes trouvées et pour chaque partition ouverte sans parent ouvert :

Pendant le démarrage de l'application, si la table des baux est vide, KCL utilise l'option ListShard API de filtrage's (le paramètre de demande ShardFilter facultatif) pour récupérer et créer des baux uniquement pour un instantané des partitions ouvertes à l'heure spécifiée par le paramètre. ShardFilter Le ShardFilter paramètre vous permet de filtrer la réponse du ListShardsAPI. La seule propriété obligatoire du paramètre ShardFilter est Type. KCLutilise la propriété Type filter et ses valeurs valides suivantes pour identifier et renvoyer un instantané des partitions ouvertes susceptibles de nécessiter de nouveaux baux :

ShardFilter est utilisé lors de la création de baux pour une table des baux vide afin d'initialiser les baux pour un instantané des partitions spécifiées à RetrievalConfig#initialPositionInStreamExtended.

Pour plus d'informations sur ShardFilter, consultez https://docs.aws.amazon.com/kinesis/latest/APIReference/API_ShardFilter.html.

Au lieu que tous les travailleurs effectuent la lease/shard synchronization to keep the lease table up to date with the latest shards in the data stream, a single elected worker leader performs the lease/shard synchronisation.

KCL2.3 utilise le paramètre de ChildShards retour de GetRecords et SubscribeToShard APIs pour effectuer la synchronisation entre bail et partition qui se produit SHARD_END pour les partitions fermées, permettant à un KCL travailleur de créer des baux uniquement pour les fragments enfants de la partition qu'il a terminé de traiter. Pour les applications grand public partagées, cette optimisation de la synchronisation entre bail et partition utilise le ChildShards paramètre de. GetRecords API Pour les applications grand public à débit dédié (fan-out amélioré), cette optimisation de la synchronisation entre location et partition utilise le paramètre de. ChildShards SubscribeToShard API Pour plus d'informations, reportez-vous aux SubscribeToShardssections GetRecords, et ChildShard.

Avec les modifications ci-dessus, le comportement de KCL passe du modèle selon lequel tous les travailleurs découvrent toutes les partitions existantes à un modèle selon lequel les travailleurs n'apprennent que les fragments enfants des fragments que chaque travailleur possède. Par conséquent, outre la synchronisation qui se produit lors des événements de démarrage et de refonte des applications grand public, elle effectue KCL désormais des analyses périodiques supplémentaires des parties/locations afin d'identifier les éventuelles lacunes dans le tableau des baux (en d'autres termes, pour en savoir plus sur toutes les nouvelles partitions) afin de garantir le traitement de la plage de hachage complète du flux de données et de créer des baux pour celles-ci si nécessaire. PeriodicShardSyncManagerest le composant chargé d'exécuter des analyses périodiques des baux et des partitions.

Pour plus d'informations sur PeriodicShardSyncManager la version KCL 2.3, consultez https://github.com/awslabs/amazon-kinesis-client/blob/master/amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/LeaseManagementConfig.java #L201 -L213.

Dans la KCL version 2.3, de nouvelles options de configuration sont disponibles pour la configuration PeriodicShardSyncManager dans LeaseManagementConfig :

De nouvelles CloudWatch mesures sont également désormais émises pour surveiller l'état de santé duPeriodicShardSyncManager. Pour de plus amples informations, veuillez consulter PeriodicShardSyncManager.

Incluant une optimisation de HierarchicalShardSyncer pour créer des baux uniquement pour une couche de partition.

Pendant le démarrage de l'application, si la table des baux est vide, KCL utilise l'option ListShard API de filtrage's (le paramètre de demande ShardFilter facultatif) pour récupérer et créer des baux uniquement pour un instantané des partitions ouvertes à l'heure spécifiée par le paramètre. ShardFilter Le ShardFilter paramètre vous permet de filtrer la réponse du ListShardsAPI. La seule propriété obligatoire du paramètre ShardFilter est Type. KCLutilise la propriété Type filter et ses valeurs valides suivantes pour identifier et renvoyer un instantané des partitions ouvertes susceptibles de nécessiter de nouveaux baux :

ShardFilter est utilisé lors de la création de baux pour une table des baux vide afin d'initialiser les baux pour un instantané des partitions spécifiées à KinesisClientLibConfiguration#initialPositionInStreamExtended.

Pour plus d'informations sur ShardFilter, consultez https://docs.aws.amazon.com/kinesis/latest/APIReference/API_ShardFilter.html.

Au lieu que tous les travailleurs effectuent la lease/shard synchronization to keep the lease table up to date with the latest shards in the data stream, a single elected worker leader performs the lease/shard synchronisation.

KCL1.14 utilise le paramètre de ChildShards retour de GetRecords et SubscribeToShard APIs pour effectuer la synchronisation entre bail et partition qui se produit SHARD_END pour les partitions fermées, permettant à un KCL travailleur de créer des baux uniquement pour les fragments enfants de la partition qu'il a terminé de traiter. Pour plus d'informations, reportez-vous GetRecordsaux sections et ChildShard.

Avec les modifications ci-dessus, le comportement de KCL passe du modèle selon lequel tous les travailleurs découvrent toutes les partitions existantes à un modèle selon lequel les travailleurs n'apprennent que les fragments enfants des fragments que chaque travailleur possède. Par conséquent, outre la synchronisation qui se produit lors des événements de démarrage et de refonte des applications grand public, elle effectue KCL désormais des analyses périodiques supplémentaires des parties/locations afin d'identifier les éventuelles lacunes dans le tableau des baux (en d'autres termes, pour en savoir plus sur toutes les nouvelles partitions) afin de garantir le traitement de la plage de hachage complète du flux de données et de créer des baux pour celles-ci si nécessaire. PeriodicShardSyncManagerest le composant chargé d'exécuter des analyses périodiques des baux et des partitions.

Lorsque KinesisClientLibConfiguration#shardSyncStrategyType est défini sur ShardSyncStrategyType.SHARD_END, PeriodicShardSync leasesRecoveryAuditorInconsistencyConfidenceThreshold est utilisé pour déterminer le seuil du nombre d'analyses consécutives révélant des lacunes dans la table des baux, après lequel une synchronisation des partitions est imposée. Lorsque KinesisClientLibConfiguration#shardSyncStrategyType est défini sur ShardSyncStrategyType.PERIODIC, leasesRecoveryAuditorInconsistencyConfidenceThreshold est ignoré.

Pour plus d'informations sur PeriodicShardSyncManager la version KCL 1.14, consultez https://github.com/awslabs/amazon-kinesis-client/blob/v1.x/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/KinesisClientLibConfiguration.java #L987 -L999.

Dans la KCL version 1.14, une nouvelle option de configuration est disponible pour configurer PeriodicShardSyncManager dans LeaseManagementConfig :

De nouvelles CloudWatch mesures sont également désormais émises pour surveiller l'état de santé duPeriodicShardSyncManager. Pour de plus amples informations, veuillez consulter PeriodicShardSyncManager.

KCLLa version 1.14 prend désormais également en charge le nettoyage différé des baux. Les baux sont supprimés de manière asynchrone par LeaseCleanupManager lorsqu'ils atteignent SHARD_END, c'est-à-dire lorsqu'une partition a dépassé la période de conservation du flux de données ou a été fermée à la suite d'une opération de repartitionnement.

De nouvelles options de configuration sont disponibles pour la configuration de LeaseCleanupManager.

Incluant une optimisation de KinesisShardSyncer pour créer des baux uniquement pour une couche de partition.

MultistreamTracker interface

Pour créer une application grand public capable de traiter plusieurs flux en même temps, vous devez implémenter une nouvelle interface appelée MultistreamTracker. Cette interface inclut la streamConfigList méthode qui renvoie la liste des flux de données et leurs configurations à traiter par l'application KCL client. Notez que les flux de données en cours de traitement peuvent être modifiés pendant l'exécution de l'application grand public. streamConfigListest appelé périodiquement par le KCL pour prendre connaissance de l'évolution des flux de données à traiter.

La streamConfigList méthode remplit la StreamConfigliste.

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;
}
                

Notez que les champs StreamIdentifier et InitialPositionInStreamExtended sont obligatoires, alors que consumerArn est facultatif. Vous ne devez fournir le consumerArn code que si vous utilisez la version KCL 2.x pour implémenter une application grand public améliorée.

Pour plus d'informations surStreamIdentifier, consultez https://github.com/awslabs/amazon-kinesis-client/blob/v2.5.8/amazon-kinesis-client/src/main/java/software/amazon/kinesis/common/StreamIdentifier.java #L129. Pour créer uneStreamIdentifier, nous vous recommandons de créer une instance multistream à partir de streamArn et streamCreationEpoch qui est disponible dans les versions 2.5.0 et ultérieures. Dans les KCL versions 2.3 et 2.4, qui ne sont pas prises en chargestreamArm, créez une instance multistream en utilisant le format. account-id:StreamName:streamCreationTimestamp Ce format sera obsolète et ne sera plus pris en charge à compter de la prochaine version majeure.

MultistreamTracker inclut également une stratégie pour supprimer les baux des anciens flux dans la table des baux (formerStreamsLeasesDeletionStrategy). Notez que la stratégie CANNOT sera modifiée pendant l'exécution de l'application grand public. Pour plus d'informations, consultez https://github.com/awslabs/amazon-kinesis-client/blob/0c5042dadf794fe988438436252a5a8fe70b6b0b//amazon-kinesis-client.java src/main/java/software/amazon/kinesis/processor/FormerStreamsLeasesDeletionStrategy

ConfigsBuilderest une classe à l'échelle de l'application que vous pouvez utiliser pour spécifier tous les paramètres de configuration KCL 2.x à utiliser lors de la création de votre KCL application grand public. ConfigsBuilderla classe prend désormais en charge l'MultistreamTrackerinterface. Vous pouvez initialiser l'un ConfigsBuilder ou l'autre avec le nom du flux de données à partir duquel les enregistrements seront consommés :

 /**
     * 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;
    }

Vous pouvez également l'initialiser ConfigsBuilder avec MultiStreamTracker si vous souhaitez implémenter une application KCL grand public qui traite plusieurs flux en même temps.

* 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;
    }                    
                

Grâce à la prise en charge multiflux mise en œuvre pour votre application KCL client, chaque ligne de la table des baux de l'application contient désormais l'ID de partition et le nom de flux des multiples flux de données traités par cette application.

Lorsque le support multiflux pour votre application KCL grand public est implémenté, la leaseKey structure est la suivante :account-id:StreamName:streamCreationTimestamp:ShardId. Par exemple, 111111111:multiStreamTest-1:12345:shardId-000000000336.