KCL을 사용한 멀티스트림 처리 - Amazon Kinesis Data Streams

기계 번역으로 제공되는 번역입니다. 제공된 번역과 원본 영어의 내용이 상충하는 경우에는 영어 버전이 우선합니다.

KCL을 사용한 멀티스트림 처리

이 섹션에서는 2개 이상의 데이터 스트림을 동시에 처리할 수 있는 KCL 소비자 애플리케이션의 생성을 가능하게 해주는 KCL의 필수 변경 사항을 설명합니다.

중요

  • 멀티스트림 처리는 KCL 2.3 이상에서만 지원됩니다.

  • 멀티스트림 처리는 비 Java 언어로 작성되어 multilangdaemon으로 실행되는 KCL 소비자는에게 지원되지 않습니다.

  • 멀티스트림 처리는 KCL 1.x의 모든 버전에서 지원되지 않습니다.

  • MultistreamTracker 인터페이스

    • 여러 스트림을 동시에 처리할 수 있는 소비자 애플리케이션을 구축하려면 MultistreamTracker라는 새 인터페이스를 구현해야 합니다. 이 인터페이스에는 KCL 소비자 애플리케이션에서 처리할 데이터 스트림 및 해당 구성 목록을 반환하는 streamConfigList 메서드가 포함되어 있습니다. 처리 중인 데이터 스트림은 소비자 애플리케이션 런타임 중에 변경될 수 있습니다. streamConfigList는 처리할 데이터 스트림의 변경 사항을 알아보기 위해 KCL에서 주기적으로 직접적으로 호출됩니다.

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

또는 동시에 여러 스트림을 처리하는 KCL 소비자 애플리케이션을 구현하려는 경우 MultiStreamTracker로 ConfigsBuilder를 초기화할 수 있습니다.


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

  • KCL 소비자 애플리케이션에 대해 멀티스트림 지원이 구현됨에 따라 이제 애플리케이션 리스 테이블의 각 행에는 이 애플리케이션이 처리하는 여러 데이터 스트림의 샤드 ID와 스트림 이름이 포함됩니다.

  • KCL 소비자 애플리케이션에 대한 멀티스트림 지원이 구현되면 leaseKey는 account-id:StreamName:streamCreationTimestamp:ShardId 구조를 취합니다. 예를 들어 111111111:multiStreamTest-1:12345:shardId-000000000336입니다.

중요

기존 KCL 소비자 애플리케이션이 하나의 데이터 스트림만 처리하도록 구성된 경우 leaseKey(리스 테이블의 파티션 키)는 샤드 ID입니다. 여러 데이터 스트림을 처리하도록 기존 KCL 소비자 애플리케이션을 재구성하면 리스 테이블이 손상됩니다. 멀티스트림을 지원하려면 leaseKey 구조가 account-id:StreamName:StreamCreationTimestamp:ShardId와 같아야 하기 때문입니다.