KCLInformazioni 1.x e 2.x

KCL1.x

Per ulteriori informazioni, consulta Sviluppa KCL consumatori 1.x

KCL2.x

Per ulteriori informazioni, consulta Sviluppa consumatori 2.x KCL

KCLapplicazione consumer: un'applicazione personalizzata utilizzando KCL e progettata per leggere ed elaborare i record dai flussi di dati.

Istanza di applicazioni consumer: le applicazioni KCL consumer sono in genere distribuite, con una o più istanze applicative in esecuzione simultanea per coordinarsi in caso di guasti e bilanciare dinamicamente il carico di elaborazione dei record di dati.

Worker: una classe di alto livello utilizzata da un'istanza di applicazioni KCL consumer per iniziare l'elaborazione dei dati.

Il worker inizializza e supervisiona varie attività, tra cui la sincronizzazione delle informazioni sulle partizioni e sui lease, il monitoraggio delle assegnazioni delle partizioni e l'elaborazione dei dati dalle partizioni. Un worker fornisce KCL le informazioni di configurazione per l'applicazione consumer, ad esempio il nome del flusso di dati i cui record di dati KCL verranno elaborati dall'applicazione consumer e le AWS credenziali necessarie per accedere a questo flusso di dati. Il worker avvia inoltre quella specifica istanza dell'applicazione KCL consumer per fornire i record di dati dal flusso di dati ai processori di record.

Lease: dati che definiscono l'associazione tra un worker e una partizione. Le applicazioni distribuite per i KCL consumatori utilizzano i contratti di locazione per suddividere l'elaborazione dei record di dati tra un parco di lavoratori. In qualsiasi momento, ogni frammento di record di dati è vincolato a un determinato lavoratore da un contratto di locazione identificato dalla variabile. leaseKey

Per impostazione predefinita, un lavoratore può detenere uno o più contratti di locazione (in base al valore della variabile maxLeasesForWorker) contemporaneamente.

Ad esempio, se si dispone di un'istanza di applicazione consumer A con worker A che elabora un flusso di dati con 4 partizioni, il worker A può detenere i lease per le partizioni 1, 2, 3 e 4 contemporaneamente. Tuttavia, se si dispone di due istanze di applicazioni consumer, A e B, con worker A e worker B, e queste istanze elaborano un flusso di dati con 4 partizioni, il worker A e il worker B non possono entrambi detenere il lease per la partizione 1 contemporaneamente. Un worker detiene il lease di una particolare partizione finché non è pronto a interrompere l'elaborazione dei record di dati della partizione o fino a quando non si verifica un guasto. Quando un worker smette di detenere il lease, un altro worker lo riprende e lo mantiene.

Per ulteriori informazioni (questi sono i KCL repository Java), vedere https://github.com/awslabs/amazon-kinesis-client/blob/v1.x/src/main/java/com/amazonaws/services/kinesis/leases/impl/Lease.java per KCL 1.x e https://github.com/awslabs/amazon-kinesis-client/blob/master/amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/Lease.java per 2.x. KCL

Tabella di leasing: una tabella Amazon DynamoDB unica che viene utilizzata per tenere traccia degli shard KDS in un flusso di dati che vengono affittati ed elaborati dai lavoratori dell'applicazione consumer. KCL La tabella di leasing deve rimanere sincronizzata (all'interno di un worker e tra tutti i lavoratori) con le informazioni più recenti sugli shard provenienti dal flusso di dati mentre l'applicazione consumer è in esecuzione. KCL Per ulteriori informazioni, consulta Utilizzate una tabella di leasing per tenere traccia degli shard elaborati dall'applicazione consumer KCL.

Processore di registrazione: la logica che definisce il modo in cui l'applicazione KCL consumer elabora i dati che riceve dai flussi di dati. In fase di esecuzione, un'istanza dell'applicazione KCL consumer crea un'istanza di un worker, che crea un'istanza di un record processor per ogni shard a cui appartiene un lease.

checkpoint: il numero di sequenza di checkpoint più recente per lo shard. Questo valore è univoco per tutte le partizioni nel flusso di dati.

checkpointSubSequenceNumero: quando si utilizza la funzione di aggregazione della Kinesis Producer Library, si tratta di un'estensione del checkpoint che tiene traccia dei record dei singoli utenti all'interno del record Kinesis.

leaseCounter: Utilizzato per il controllo delle versioni del leasing in modo che i lavoratori possano rilevare che il contratto di locazione è stato preso da un altro lavoratore.

leaseKey: un identificatore univoco per un contratto di locazione. Ogni lease è specifico di una partizione nel flusso di dati ed è detenuto da un worker alla volta.

leaseOwner: Il lavoratore titolare del contratto di locazione.

ownerSwitchesSinceCheckpoint: quante volte questo contratto di locazione ha cambiato lavoratori dall'ultima volta che è stato scritto un checkpoint.

parentShardId: Utilizzato per garantire che lo shard principale sia completamente elaborato prima che inizi l'elaborazione sui frammenti secondari. In questo modo, ci si assicura che i record siano elaborati nello stesso ordine in cui sono stati introdotti nel flusso.

hashrange: utilizzato dalla PeriodicShardSyncManager per eseguire sincronizzazioni periodiche per trovare le partizioni mancanti nella tabella di lease e creare lease per esse, se necessario.

childshards: utilizzato da LeaseCleanupManager per esaminare lo stato di elaborazione della partizione secondaria e decidere se la partizione principale può essere eliminata dalla tabella di lease.

shardID: l'ID della partizione.

nome del flusso: l'identificatore del flusso di dati nel seguente formato: account-id:StreamName:streamCreationTimestamp.

Recupera tutte le partizioni per i dati elaborati dal flusso

Recupera tutte i lease delle partizioni dalla tabella di lease

Filtra ogni partizione aperta che non ha un lease nella tabella di lease

Itera su tutte le partizioni aperte trovate e per ogni partizione aperta senza un elemento principale aperto:

Durante l'avvio dell'applicazione, se la tabella di lease è vuota, KCL utilizza l'opzione di filtro ListShard API dell'applicazione (il parametro di richiesta ShardFilter opzionale) per recuperare e creare leasing solo per un'istantanea degli shard aperti nel momento specificato dal parametro. ShardFilter Il ShardFilter parametro consente di filtrare la risposta di. ListShards API L'unica proprietà richiesta del parametro ShardFilter è Type. KCLutilizza la proprietà Type filter e i seguenti valori validi per identificare e restituire un'istantanea degli shard aperti che potrebbero richiedere nuovi leasing:

ShardFilter viene utilizzato durante la creazione di lease per una tabella di lease vuota per inizializzare i lease per uno snapshot delle partizioni specificate in RetrievalConfig#initialPositionInStreamExtended.

Per ulteriori informazioni su ShardFilter, consulta https://docs.aws.amazon.com/kinesis/latest/APIReference/API_ShardFilter.html.

Invece che tutti i lavoratori eseguano la sincronizzazione. 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

KCL2.3 utilizza il parametro ChildShards return di GetRecords and the SubscribeToShard APIs per eseguire la sincronizzazione lease/shard che avviene per i frammenti chiusi, consentendo a un KCL lavoratore di creare leasing solo SHARD_END per i frammenti secondari dello shard che ha terminato l'elaborazione. Per la condivisione tra le applicazioni consumer, questa ottimizzazione della sincronizzazione lease/shard utilizza il parametro di. ChildShards GetRecords API Per le applicazioni consumer con throughput dedicato (fan-out avanzato), questa ottimizzazione della sincronizzazione lease/shard utilizza il parametro di. ChildShards SubscribeToShard API Per ulteriori informazioni, vedere, e. GetRecordsSubscribeToShardsChildShard

Con le modifiche di cui sopra, il comportamento di KCL sta passando dal modello di tutti i lavoratori che imparano a conoscere tutti i frammenti esistenti al modello dei lavoratori che apprendono solo i bambini, frammenti dei frammenti di frammenti di proprietà di ciascun lavoratore. Pertanto, oltre alla sincronizzazione che avviene durante gli eventi di bootstraping e reshard delle applicazioni consumer, KCL ora esegue anche ulteriori scansioni periodiche degli shard/lease per identificare eventuali potenziali buchi nella tabella di lease (in altre parole, per conoscere tutti i nuovi shard) per garantire l'elaborazione dell'intervallo hash completo del flusso di dati e creare contratti di leasing, se necessario. PeriodicShardSyncManagerè il componente responsabile dell'esecuzione di scansioni periodiche di lease/shard.

Per ulteriori informazioni su PeriodicShardSyncManager in KCL 2.3, vedere/.java #L201 -L213. https://github.com/awslabs/ amazon-kinesis-client blob/master/amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/LeaseManagementConfig

Nella versione KCL 2.3, sono disponibili nuove opzioni di configurazione da configurare in: PeriodicShardSyncManager LeaseManagementConfig

Ora vengono inoltre emesse nuove CloudWatch metriche per monitorare lo stato di. PeriodicShardSyncManager Per ulteriori informazioni, consulta PeriodicShardSyncManager.

Inclusa un'ottimizzazione a HierarchicalShardSyncer per creare lease solo per un livello di partizioni.

Durante l'avvio dell'applicazione, se la tabella di lease è vuota, KCL utilizza l'opzione di filtro ListShard API dell'applicazione (il parametro di richiesta ShardFilter opzionale) per recuperare e creare leasing solo per un'istantanea degli shard aperti nel momento specificato dal parametro. ShardFilter Il ShardFilter parametro consente di filtrare la risposta di. ListShards API L'unica proprietà richiesta del parametro ShardFilter è Type. KCLutilizza la proprietà Type filter e i seguenti valori validi per identificare e restituire un'istantanea degli shard aperti che potrebbero richiedere nuovi leasing:

ShardFilter viene utilizzato durante la creazione di lease per una tabella di lease vuota per inizializzare i lease per uno snapshot delle partizioni specificate in KinesisClientLibConfiguration#initialPositionInStreamExtended.

Per ulteriori informazioni su ShardFilter, consulta https://docs.aws.amazon.com/kinesis/latest/APIReference/API_ShardFilter.html.

Invece che tutti i lavoratori eseguano la sincronizzazione. 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

KCL1.14 utilizza il parametro ChildShards return di GetRecords and the SubscribeToShard APIs per eseguire la sincronizzazione lease/shard che avviene per i frammenti chiusi, consentendo a un KCL lavoratore di creare leasing solo SHARD_END per i frammenti secondari dello shard che ha terminato l'elaborazione. Per ulteriori informazioni, vedere e. GetRecordsChildShard

Con le modifiche di cui sopra, il comportamento di KCL sta passando dal modello di tutti i lavoratori che imparano a conoscere tutti i frammenti esistenti al modello dei lavoratori che apprendono solo i bambini, frammenti dei frammenti di frammenti di proprietà di ciascun lavoratore. Pertanto, oltre alla sincronizzazione che avviene durante gli eventi di bootstraping e reshard delle applicazioni consumer, KCL ora esegue anche ulteriori scansioni periodiche degli shard/lease per identificare eventuali potenziali buchi nella tabella di lease (in altre parole, per conoscere tutti i nuovi shard) per garantire l'elaborazione dell'intervallo hash completo del flusso di dati e creare contratti di leasing, se necessario. PeriodicShardSyncManagerè il componente responsabile dell'esecuzione di scansioni periodiche di lease/shard.

Quando KinesisClientLibConfiguration#shardSyncStrategyType è impostato su ShardSyncStrategyType.SHARD_END, PeriodicShardSync leasesRecoveryAuditorInconsistencyConfidenceThreshold viene utilizzato per determinare la soglia per il numero di scansioni consecutive contenenti buchi nella tabella di lease, dopodiché imporre la sincronizzazione delle partizioni. Quando KinesisClientLibConfiguration#shardSyncStrategyType è impostato su ShardSyncStrategyType.PERIODIC, leasesRecoveryAuditorInconsistencyConfidenceThreshold viene ignorato.

Per ulteriori informazioni sulla versione KCL 1.14, vedere/PeriodicShardSyncManager.java #L987 -L999. https://github.com/awslabs/ amazon-kinesis-client blob/v1.x/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/KinesisClientLibConfiguration

Nella versione KCL 1.14, è disponibile una nuova opzione di configurazione da configurare in: PeriodicShardSyncManager LeaseManagementConfig

Ora vengono inoltre emesse nuove CloudWatch metriche per monitorare lo stato di. PeriodicShardSyncManager Per ulteriori informazioni, consulta PeriodicShardSyncManager.

KCL1.14 ora supporta anche la pulizia del leasing differito. I lease vengono eliminati in modo asincrono da LeaseCleanupManager quando viene raggiunto SHARD_END o quando una partizione è scaduta, superando il periodo di conservazione del flusso di dati o è stata chiusa a seguito di un'operazione di ripartizionamento.

Sono disponibili nuove opzioni di configurazione per configurare LeaseCleanupManager.

Inclusa un'ottimizzazione a KinesisShardSyncer per creare lease solo per un livello di partizioni.

MultistreamTracker interfaccia

Per creare un'applicazione consumer in grado di elaborare più flussi contemporaneamente, è necessario implementare una nuova interfaccia denominata MultistreamTracker. Questa interfaccia include il streamConfigList metodo che restituisce l'elenco dei flussi di dati e le relative configurazioni che devono essere elaborati dall'KCLapplicazione consumer. Si noti che i flussi di dati in fase di elaborazione possono essere modificati durante il runtime dell'applicazione consumer. streamConfigListviene chiamato periodicamente dal KCL per conoscere le modifiche nei flussi di dati da elaborare.

Il streamConfigList metodo compila l'elenco. 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;
}
                

Nota che StreamIdentifier e InitialPositionInStreamExtended sono obbligatori, mentre consumerArn è facoltativo. È necessario fornire il comando consumerArn solo se si utilizza la versione KCL 2.x per implementare un'applicazione consumer fan-out avanzata.

Per ulteriori informazioni suStreamIdentifier, vedere https://github.com/awslabs/amazon-kinesis-client/blob/v2.5.8/amazon-kinesis-client/src/main/java/software/amazon/kinesis/common/StreamIdentifier.java #L129. Per creare unStreamIdentifier, ti consigliamo di creare un'istanza multistream da streamArn and the disponibile nella versione streamCreationEpoch 2.5.0 e successive. Nelle KCL versioni 2.3 e v2.4, che non supportanostreamArm, crea un'istanza multistream utilizzando il formato. account-id:StreamName:streamCreationTimestamp Questo formato sarà obsoleto e non sarà più supportato a partire dalla prossima versione principale.

MultistreamTracker include anche una strategia per eliminare i lease di vecchi flussi nella tabella dei lease (formerStreamsLeasesDeletionStrategy). Si noti che la strategia viene CANNOT modificata durante il runtime dell'applicazione consumer. Per ulteriori informazioni, consultate https://github.com/awslabs/amazon-kinesis-client/blob/0c5042dadf794fe988438436252a5a8fe70b6b0b//amazon-kinesis-client.java src/main/java/software/amazon/kinesis/processor/FormerStreamsLeasesDeletionStrategy

ConfigsBuilderè una classe a livello di applicazione che è possibile utilizzare per specificare tutte le impostazioni di configurazione 2.x da utilizzare durante la creazione di un'applicazione consumer. KCL KCL ConfigsBuilderla classe ora supporta l'interfaccia. MultistreamTracker Puoi inizializzarli ConfigsBuilder entrambi con il nome dell'unico flusso di dati da cui consumare i record da:

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

Oppure puoi inizializzare ConfigsBuilder con MultiStreamTracker se desideri implementare un'applicazione KCL consumer che elabori più flussi contemporaneamente.

* 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 il supporto multistream implementato per l'applicazione KCL consumer, ogni riga della tabella di lease dell'applicazione ora contiene l'ID dello shard e il nome del flusso dei molteplici flussi di dati elaborati dall'applicazione.

Quando viene implementato il supporto multistream per la tua applicazione KCL consumer, assume la seguente struttura:. leaseKey account-id:StreamName:streamCreationTimestamp:ShardId Ad esempio 111111111:multiStreamTest-1:12345:shardId-000000000336.