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ù flussi 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. Si noti 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 inizializzare ConfigsBuilder con MultiStreamTracker se desideri implementare un'applicazione consumer KCL che elabora 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 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 del flusso dei molteplici flussi di dati elaborati da questa applicazione.

  • 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 la tua applicazione consumer KCL esistente è configurata per elaborare solo un flusso di dati, leaseKey (che è la chiave di partizione per la tabella di leasing) è lo shard ID. Se riconfigurate un'applicazione consumer KCL esistente per elaborare più flussi di dati, la tabella di leasing si interrompe, perché la leaseKey struttura deve essere la seguente: per supportare il multi-stream. account-id:StreamName:StreamCreationTimestamp:ShardId