KCLInformationen zu 1.x und 2.x

KCL1.x

Weitere Informationen finden Sie unter Entwickeln Sie KCL 1.x-Verbraucher

KCL2.x

Weitere Informationen finden Sie unter Entwickeln Sie KCL 2.x-Verbraucher

KCLAnwendung für Privatanwender — eine Anwendung, die speziell für das Lesen KCL und Verarbeiten von Datensätzen aus Datenströmen entwickelt wurde.

Anwendungsinstanz für KCL Privatanwender — Anwendungen für Privatanwender sind in der Regel verteilt, wobei eine oder mehrere Anwendungsinstanzen gleichzeitig ausgeführt werden, um Fehler zu koordinieren und einen dynamischen Lastenausgleich bei der Verarbeitung von Datensätzen vorzunehmen.

Worker — eine übergeordnete Klasse, die eine Anwendungsinstanz für KCL Privatanwender verwendet, um mit der Datenverarbeitung zu beginnen.

Der Worker initialisiert und überwacht verschiedene Aufgaben, darunter das Synchronisieren von Shard- und Leasing-Informationen, das Verfolgen von Shard-Zuweisungen und das Verarbeiten von Daten aus den Shards. Ein Worker stellt KCL die Konfigurationsinformationen für die Verbraucheranwendung bereit, z. B. den Namen des Datenstroms, dessen Datensätze diese KCL Verbraucheranwendung verarbeiten wird, und die AWS Anmeldeinformationen, die für den Zugriff auf diesen Datenstrom erforderlich sind. Der Worker startet außerdem die spezifische Instanz der KCL Verbraucheranwendung, um Datensätze aus dem Datenstrom an die Datensatzprozessoren weiterzuleiten.

Lease – Daten, die die Bindung zwischen einem Worker und einem Shard definieren. Verteilte KCL Verbraucheranwendungen verwenden Leases, um die Verarbeitung von Datensätzen auf eine ganze Flotte von Mitarbeitern zu verteilen. Zu einem bestimmten Zeitpunkt ist jede Datenmenge durch einen durch die Variable identifizierten Leasingvertrag an einen bestimmten Mitarbeiter gebunden. leaseKey

Standardmäßig kann ein Arbeitnehmer einen oder mehrere Mietverträge (abhängig vom Wert der Variablen maxLeasesForWorker) gleichzeitig abschließen.

Wenn Sie beispielsweise eine Konsumentenanwendungs-Instance A mit Worker A haben, die einen Datenstrom mit 4 Shards verarbeitet, kann Worker A Leases für die Shards 1, 2, 3 und 4 gleichzeitig halten. Wenn Sie jedoch zwei Konsumentenanwendungs-Instances haben: A und B mit Worker A und Worker B und diese Instances einen Datenstrom mit 4 Shards verarbeiten, können Worker A und Worker B nicht gleichzeitig den Lease für Shard 1 halten. Ein Worker hält den Lease für einen bestimmten Shard, bis er bereit ist, die Verarbeitung der Datensätze dieses Shards zu beenden, oder bis er ausfällt. Wenn ein Worker den Lease beendet, nimmt ein anderer Worker den Lease auf und hält ihn.

Weitere Informationen (dies sind die KCL Java-Repositorys) finden Sie unter https://github.com/awslabs/amazon-kinesis-client/blob/v1.x/src/main/java/com/amazonaws/services/kinesis/leases/impl/Lease.java für KCL 1.x und https://github.com/awslabs/amazon-kinesis-client/blob/master/amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/Lease.java für 2.x. KCL

Leasing-Tabelle — eine einzigartige Amazon DynamoDB-Tabelle, die verwendet wird, um die Shards in einem KDS Datenstrom zu verfolgen, die von den Mitarbeitern der Verbraucheranwendung geleast und verarbeitet werden. KCL Die Leasing-Tabelle muss (innerhalb eines Workers und für alle Mitarbeiter) mit den neuesten Shard-Informationen aus dem Datenstrom synchron bleiben, während die Consumer-Anwendung ausgeführt wirdKCL. Weitere Informationen finden Sie unter Verwenden Sie eine Leasetabelle, um nachzuverfolgen, welche Shards von der Consumer-Anwendung verarbeitet wurden KCL.

Record Processor — die Logik, die definiert, wie Ihre KCL Verbraucheranwendung die Daten verarbeitet, die sie aus den Datenströmen erhält. Zur Laufzeit instanziiert eine KCL Consumer-Anwendungsinstanz einen Worker, und dieser Worker instanziiert einen Datensatzprozessor für jeden Shard, für den er eine Lease hält.

checkpoint: Die letzte Prüfpunkt-Sequenznummer des Shards. Dieser Wert ist für alle Shards im Datenstrom eindeutig.

checkpointSubSequenceNummer: Wenn Sie die Aggregationsfunktion der Kinesis Producer Library verwenden, ist dies eine Erweiterung des Checkpoints, mit der einzelne Benutzerdatensätze innerhalb des Kinesis-Datensatzes verfolgt werden.

leaseCounter: Wird für die Versionierung von Leasing-Verträgen verwendet, sodass Mitarbeiter erkennen können, dass ihr Leasing von einem anderen Mitarbeiter angenommen wurde.

leaseKey: Eine eindeutige Kennung für einen Mietvertrag. Jeder Lease gilt für einen bestimmten Shard im Datenstrom und wird immer nur von einem Worker gehalten.

leaseOwner: Der Arbeitnehmer, der diesen Mietvertrag innehat.

ownerSwitchesSinceCheckpoint: Wie oft hat dieser Mietvertrag die Mitarbeiter gewechselt, seit ein Checkpoint das letzte Mal geschrieben wurde.

parentShardId: Wird verwendet, um sicherzustellen, dass der übergeordnete Shard vollständig verarbeitet ist, bevor die Verarbeitung der untergeordneten Shards beginnt. So wird sichergestellt, dass Datensätze in der gleichen Reihenfolge verarbeitet werden, in der sie in den Stream eingespeist wurden.

hashrange: Wird von PeriodicShardSyncManager verwendet, um regelmäßige Synchronisierungen durchzuführen, um fehlende Shards in der Leasetabelle zu finden und bei Bedarf Leases für sie zu erstellen.

childshards: Wird von LeaseCleanupManager verwendet, um den Verarbeitungsstatus des untergeordneten Shards zu überprüfen und zu entscheiden, ob der übergeordnete Shard aus der Leasetabelle gelöscht werden kann.

shardID: Die ID des Shards.

Streamname Die Kennung des Datenstroms im folgenden Format: account-id:StreamName:streamCreationTimestamp.

Ruft alle Shards für Daten des Streams ab, der gerade verarbeitet wird

Ruft alle Shard-Leases aus der Leasetabelle ab

Filtert jeden offenen Shard heraus, für den es in der Leasetabelle keinen Lease gibt

Iteriert über alle gefundenen offenen Shards und für jeden offenen Shard ohne offenes übergeordnetes Element:

Wenn die Leasetabelle beim Bootstraping der Anwendung leer ist, KCL verwendet sie die Filteroption ListShard API von (den ShardFilter optionalen Anforderungsparameter), um Leases nur für einen Snapshot von Shards abzurufen und zu erstellen, die zu dem durch den Parameter angegebenen Zeitpunkt geöffnet sind. ShardFilter Mit dem ShardFilter Parameter können Sie die Antwort von herausfiltern. ListShards API Die einzige erforderliche Eigenschaft des ShardFilter-Parameters ist Type. KCLverwendet die Eigenschaft Type filter und die folgenden gültigen Werte, um offene Shards, für die möglicherweise neue Leases erforderlich sind, zu identifizieren und eine Momentaufnahme zurückzugeben:

ShardFilter wird verwendet, wenn Leases für eine leere Leasetabelle erstellt werden, um Leases für einen Snapshot von Shards zu initialisieren, die unter RetrievalConfig#initialPositionInStreamExtended angegeben sind.

Mehr über ShardFilter erfahren Sie unter https://docs.aws.amazon.com/kinesis/latest/APIReference/API_ShardFilter.html.

Anstatt dass alle Worker die 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 durchführen.

KCL2.3 verwendet den ChildShards Rückgabeparameter von GetRecords und, SubscribeToShard APIs um die Leasing-/Shard-Synchronisierung durchzuführen, die bei SHARD_END geschlossenen Shards stattfindet, sodass ein KCL Worker nur Leases für die untergeordneten Shards des Shards erstellen kann, dessen Verarbeitung abgeschlossen ist. Diese Optimierung der Leasing-/Shard-Synchronisierung verwendet bei der gemeinsamen Nutzung von Consumer-Anwendungen den Parameter von. ChildShards GetRecords API Für Verbraucheranwendungen mit dediziertem Durchsatz (erweiterter Fan-Out) verwendet diese Optimierung der Lease-/Shard-Synchronisierung den Parameter von. ChildShards SubscribeToShard API Weitere Informationen finden Sie unter GetRecords, und. SubscribeToShardsChildShard

Mit den oben genannten Änderungen geht das Verhalten von von KCL weg von dem Modell, dass alle Mitarbeiter über alle vorhandenen Shards Bescheid wissen, hin zu dem Modell, dass Arbeitnehmer nur über die untergeordneten Scherben der Shards, die jeder Arbeiter besitzt, lernen. Daher werden neben der Synchronisation, die beim Bootstraping von Verbraucheranwendungen und bei Reshard-Ereignissen stattfindet, KCL jetzt auch zusätzliche regelmäßige Shard-/Lease-Scans durchgeführt, um potenzielle Lücken in der Leasetabelle zu identifizieren (mit anderen Worten, um mehr über alle neuen Shards zu erfahren), um sicherzustellen, dass der gesamte Hash-Bereich des Datenstroms verarbeitet wird, und um bei Bedarf Leases für sie zu erstellen. PeriodicShardSyncManagerist die Komponente, die für die regelmäßige Ausführung von Lease-/Shard-Scans verantwortlich ist.

Weitere Informationen zu PeriodicShardSyncManager Version KCL 2.3 finden Sie unter https://github.com/awslabs/amazon-kinesis-client/blob/master/amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/LeaseManagementConfig.java #L201 -L213.

In KCL 2.3 stehen neue Konfigurationsoptionen zur Konfiguration zur Verfügung in: PeriodicShardSyncManager LeaseManagementConfig

Außerdem werden jetzt neue CloudWatch Messwerte ausgegeben, um den Zustand von zu überwachenPeriodicShardSyncManager. Weitere Informationen finden Sie unter PeriodicShardSyncManager.

Einschließlich einer Optimierung auf HierarchicalShardSyncer, um nur Leases für eine Shard-Ebene zu erstellen.

Wenn die Leasetabelle beim Bootstraping der Anwendung leer ist, KCL verwendet sie die Filteroption ListShard API von (den ShardFilter optionalen Anforderungsparameter), um Leases nur für einen Snapshot von Shards abzurufen und zu erstellen, die zu dem durch den Parameter angegebenen Zeitpunkt geöffnet sind. ShardFilter Mit dem ShardFilter Parameter können Sie die Antwort von herausfiltern. ListShards API Die einzige erforderliche Eigenschaft des ShardFilter-Parameters ist Type. KCLverwendet die Eigenschaft Type filter und die folgenden gültigen Werte, um offene Shards, für die möglicherweise neue Leases erforderlich sind, zu identifizieren und eine Momentaufnahme zurückzugeben:

ShardFilter wird verwendet, wenn Leases für eine leere Leasetabelle erstellt werden, um Leases für einen Snapshot von Shards zu initialisieren, die unter KinesisClientLibConfiguration#initialPositionInStreamExtended angegeben sind.

Mehr über ShardFilter erfahren Sie unter https://docs.aws.amazon.com/kinesis/latest/APIReference/API_ShardFilter.html.

Anstatt dass alle Worker die 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 durchführen.

KCL1.14 verwendet den ChildShards Rückgabeparameter von GetRecords und, SubscribeToShard APIs um die Leasing-/Shard-Synchronisierung durchzuführen, die bei geschlossenen Shards stattfindet, sodass ein KCL Worker nur Leases SHARD_END für die untergeordneten Shards des Shards erstellen kann, dessen Verarbeitung abgeschlossen ist. GetRecordsWeitere ChildShardInformationen finden Sie unter und.

Mit den oben genannten Änderungen KCL geht das Verhalten von weg von dem Modell, dass alle Arbeitnehmer über alle vorhandenen Scherben Bescheid wissen, hin zu dem Modell, dass die Arbeitnehmer nur über die untergeordneten Scherben der Scherben, die jeder Arbeiter besitzt, lernen. Daher werden neben der Synchronisation, die beim Bootstraping von Verbraucheranwendungen und bei Reshard-Ereignissen stattfindet, KCL jetzt auch zusätzliche regelmäßige Shard-/Lease-Scans durchgeführt, um potenzielle Lücken in der Leasetabelle zu identifizieren (mit anderen Worten, um mehr über alle neuen Shards zu erfahren), um sicherzustellen, dass der gesamte Hash-Bereich des Datenstroms verarbeitet wird, und um bei Bedarf Leases für sie zu erstellen. PeriodicShardSyncManagerist die Komponente, die für die regelmäßige Ausführung von Lease-/Shard-Scans verantwortlich ist.

Wenn KinesisClientLibConfiguration#shardSyncStrategyType auf ShardSyncStrategyType.SHARD_END gesetzt ist, wird PeriodicShardSync leasesRecoveryAuditorInconsistencyConfidenceThreshold verwendet, um den Schwellenwert für die Anzahl der aufeinanderfolgenden Scans mit Lücken in der Leasetabelle zu bestimmen, nach dem eine Shard-Synchronisierung erzwungen werden soll. Wenn KinesisClientLibConfiguration#shardSyncStrategyType auf ShardSyncStrategyType.PERIODIC gesetzt ist, wird leasesRecoveryAuditorInconsistencyConfidenceThreshold ignoriert.

Weitere Informationen zu PeriodicShardSyncManager Version KCL 1.14 finden Sie unter https://github.com/awslabs/amazon-kinesis-client/blob/v1.x/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/KinesisClientLibConfiguration.java #L987 -L999.

In KCL 1.14 steht eine neue Konfigurationsoption zur Konfiguration zur Verfügung in: PeriodicShardSyncManager LeaseManagementConfig

Neue CloudWatch Metriken werden jetzt auch ausgegeben, um den Zustand von zu überwachen. PeriodicShardSyncManager Weitere Informationen finden Sie unter PeriodicShardSyncManager.

KCL1.14 unterstützt jetzt auch die verzögerte Leasingbereinigung. Leases werden asynchron von LeaseCleanupManager bei Erreichen von SHARD_END gelöscht, wenn ein Shard entweder die Aufbewahrungsfrist des Datenstroms überschritten hat oder wenn er aufgrund eines Resharding-Vorgangs geschlossen wurde.

Es stehen neue Konfigurationsoptionen zur Verfügung, um LeaseCleanupManager zu konfigurieren.

Einschließlich einer Optimierung auf KinesisShardSyncer, um nur Leases für eine Shard-Ebene zu erstellen.

MultistreamTracker Schnittstelle

Um eine Verbraucheranwendung zu erstellen, die mehrere Streams gleichzeitig verarbeiten kann, müssen Sie eine neue Schnittstelle namens implementieren MultistreamTracker. Diese Schnittstelle enthält die streamConfigList Methode, die die Liste der Datenströme und ihrer Konfigurationen zurückgibt, die von der KCL Verbraucheranwendung verarbeitet werden sollen. Beachten Sie, dass die Datenströme, die verarbeitet werden, während der Laufzeit der Verbraucheranwendung geändert werden können. streamConfigListwird regelmäßig von der aufgerufenKCL, um mehr über die Änderungen der zu verarbeitenden Datenströme zu erfahren.

Die streamConfigList Methode füllt die StreamConfigListe auf.

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

Beachten Sie, dass es sich bei den Feldern StreamIdentifier und InitialPositionInStreamExtended um Pflichtfelder handelt, während consumerArn optional ist. Sie müssen das consumerArn nur angeben, wenn Sie KCL 2.x verwenden, um eine erweiterte Fan-Out-Consumer-Anwendung zu implementieren.

Weitere Informationen StreamIdentifier dazu finden Sie unter https://github.com/awslabs/amazon-kinesis-client/blob/v2.5.8/amazon-kinesis-client/src/main/java/software/amazon/kinesis/common/StreamIdentifier.java #L129. Um eine zu erstellenStreamIdentifier, empfehlen wir, dass Sie eine Multistream-Instanz aus der streamArn und der erstellenstreamCreationEpoch, die in Version 2.5.0 und höher verfügbar ist. Erstellen Sie in KCL Version 2.3 und Version 2.4, die dies nicht unterstützenstreamArm, eine Multistream-Instance mithilfe des folgenden Formats. account-id:StreamName:streamCreationTimestamp Dieses Format ist veraltet und wird ab der nächsten Hauptversion nicht mehr unterstützt.

MultistreamTracker beinhaltet auch eine Strategie zum Löschen von Leases alter Streams in der Leasetabelle (formerStreamsLeasesDeletionStrategy). Beachten Sie, dass die Strategie CANNOT während der Laufzeit der Verbraucheranwendung geändert wird. Weitere Informationen finden Sie unter https://github.com/awslabs/amazon-kinesis-client/blob/0c5042dadf794fe988438436252a5a8fe70b6b0b//amazon-kinesis-client.java src/main/java/software/amazon/kinesis/processor/FormerStreamsLeasesDeletionStrategy

ConfigsBuilderist eine anwendungsweite Klasse, mit der Sie alle 2.x-Konfigurationseinstellungen angeben können, die beim Erstellen Ihrer Verbraucheranwendung verwendet werden sollen. KCL KCL ConfigsBuilderDie Klasse unterstützt jetzt die Schnittstelle. MultistreamTracker Sie können ConfigsBuilder entweder mit dem Namen des einen Datenstroms initialisieren, aus dem Datensätze abgerufen werden sollen:

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

Oder Sie können ConfigsBuilder mit initialisieren, MultiStreamTracker wenn Sie eine KCL Verbraucheranwendung implementieren möchten, die mehrere Streams gleichzeitig verarbeitet.

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

Da die Multistream-Unterstützung für Ihre KCL Consumer-Anwendung implementiert ist, enthält jede Zeile der Leasetabelle der Anwendung jetzt die Shard-ID und den Stream-Namen der mehreren Datenströme, die diese Anwendung verarbeitet.

Wenn die Multistream-Unterstützung für Ihre KCL Verbraucheranwendung implementiert ist, hat sie die leaseKey folgende Struktur:. account-id:StreamName:streamCreationTimestamp:ShardId Beispiel, 111111111:multiStreamTest-1:12345:shardId-000000000336.