View a markdown version of this page

Multi-stream elaborazione con KCL - Flusso di dati Amazon Kinesis

Le traduzioni sono generate tramite traduzione automatica. In caso di conflitto tra il contenuto di una traduzione e la versione originale in Inglese, quest'ultima prevarrà.

Multi-stream elaborazione con KCL

Questa sezione descrive le modifiche richieste in KCL che consentono di creare applicazioni consumer KCL in grado di elaborare più di un flusso di dati contemporaneamente.

Importante

  • Multi-stream l'elaborazione è supportata solo in KCL 2.3 o versioni successive.

  • Multi-stream l'elaborazione non è supportata per i consumatori KCL scritti in linguaggi non Java che funzionano con. multilangdaemon

  • Multi-stream l'elaborazione non è supportata in nessuna versione di KCL 1.x.

  • MultistreamTracker interfaccia

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

    • Compila streamConfigList 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;
}

Oppure puoi inizializzarlo ConfigsBuilder con MultiStreamTracker se desideri implementare un'applicazione consumer KCL che elabori più stream 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 multi-stream implementato per la tua applicazione consumer KCL, ogni riga della tabella di lease dell'applicazione ora contiene l'ID dello shard e il nome dello stream dei molteplici flussi di dati che questa applicazione elabora.

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

Importante

Quando l'applicazione consumer KCL esistente è configurata per elaborare un solo flusso di dati, la leaseKey (che è la chiave di partizione per la tabella di leasing) è l'ID dello shard. Se riconfiguri un'applicazione consumer KCL esistente per elaborare più flussi di dati, la tabella di leasing viene interrotta, perché la leaseKey struttura deve essere la seguente: supportare il multi-stream. account-id:StreamName:StreamCreationTimestamp:ShardId