Kinesis 클라이언트 라이브러리 사용

데이터 스트림에 연결합니다

데이터 스트림 내 샤드를 열거합니다.

임대를 사용하여 샤드 연결을 해당 워커와 조정합니다.

관리하는 모든 샤드의 레코드 프로세서를 인스턴스화합니다

데이터 스트림에서 데이터 레코드를 가져옵니다.

해당하는 레코드 프로세서로 레코드를 푸시합니다

처리된 레코드에 대해 체크포인트를 수행합니다

워커 인스턴스 수가 변경되거나 데이터 스트림이 리샤딩(샤드 분할 또는 병합)될 때 샤드-워커 연결(리스)의 균형을 유지합니다.

KCL1.x

자세한 내용은 KCL1.x 소비자 개발 단원을 참조하세요.

KCL2.x

자세한 내용은 KCL2.x 소비자 개발 단원을 참조하세요.

KCL소비자 애플리케이션 — 데이터 스트림을 사용하여 맞춤 KCL 구축되고 데이터 스트림에서 레코드를 읽고 처리하도록 설계된 애플리케이션입니다.

소비자 애플리케이션 인스턴스 - KCL 소비자 애플리케이션은 일반적으로 분산되어 있으며, 장애 발생 시 이를 조정하고 데이터 레코드 처리의 부하를 동적으로 분산하기 위해 하나 이상의 애플리케이션 인스턴스가 동시에 실행됩니다.

작업자 — KCL 소비자 애플리케이션 인스턴스가 데이터 처리를 시작하는 데 사용하는 상위 수준 클래스입니다.

워커는 샤드 및 리스 정보 동기화, 샤드 할당 추적, 샤드의 데이터 처리 등 다양한 작업을 초기화하고 감독합니다. 워커는 소비자 애플리케이션에 대한 구성 정보 (예: 이 소비자 애플리케이션이 처리할 데이터 레코드의 데이터 스트림 이름, 이 KCL 데이터 스트림에 액세스하는 데 필요한 AWS 자격 증명) 를 제공합니다KCL. 또한 워커는 해당 특정 KCL 소비자 애플리케이션 인스턴스를 킥시작하여 데이터 스트림의 데이터 레코드를 레코드 프로세서로 전달합니다.

리스 - 워커와 샤드 간의 바인딩을 정의하는 데이터입니다. 분산된 KCL 소비자 애플리케이션은 임대를 사용하여 여러 작업자 간에 데이터 레코드 처리를 분할합니다. 언제든지 각 데이터 레코드 분할은 변수로 식별되는 임대 계약을 통해 특정 작업자에게 바인딩됩니다. leaseKey

기본적으로 작업자는 작업자 변수 값에 따라 하나 이상의 임대를 동시에 보유할 수 있습니다. maxLeasesFor

예를 들어, 소비자 애플리케이션 인스턴스 A에 워커 A가 있고 이 인스턴스가 4개의 샤드로 구성된 데이터 스트림을 처리하는 경우 워커 A는 샤드 1, 2, 3, 4에 대한 리스를 동시에 보유할 수 있습니다. 그러나 A와 B라는 2개의 소비자 애플리케이션 인스턴스에 각각 워커 A와 워커 B가 있고 이들 인스턴스가 4개의 샤드로 구성된 데이터 스트림을 처리하는 경우 워커 A와 워커 B는 샤드 1에 대한 리스를 동시에 보유할 수 없습니다. 한 워커가 특정 샤드의 데이터 레코드 처리를 중지할 준비가 되거나 실패할 때까지 이 샤드에 대한 리스를 보유합니다. 한 워커가 리스 보유를 중지하면 다른 워커가 리스를 인수하여 보유합니다.

자바 KCL 리포지토리에 대한 자세한 내용은 1.x의 경우 https://github.com/awslabs/amazon-kinesis-client/blob/v1.x/src/main/java/com/AmazonAWS/services/kinesis/leases/impl/lease.java를, 2.x의 경우 /blob/master/ /src/main/java/software/amazon/kinesis/leases/Lease.java 를 참조하십시오. KCL https://github.com/awslabs/ amazon-kinesis-client amazon-kinesis-client KCL

임대 테이블 - 소비자 애플리케이션 작업자가 임대하고 처리 중인 데이터 스트림의 샤드를 추적하는 데 사용되는 고유한 Amazon DynamoDB 테이블입니다. KDS KCL 소비자 애플리케이션이 실행되는 동안 임대 테이블은 데이터 스트림의 최신 샤드 정보와 동기화 상태를 유지해야 합니다 (작업자 내부 및 모든 작업자 간에). KCL 자세한 내용은 임대 테이블을 사용하여 소비자 애플리케이션에서 처리한 샤드를 추적할 수 있습니다. KCL 단원을 참조하십시오.

레코드 프로세서 — KCL 소비자 애플리케이션이 데이터 스트림에서 가져온 데이터를 처리하는 방법을 정의하는 로직입니다. 런타임 시 KCL 소비자 애플리케이션 인스턴스는 워커를 인스턴스화하고, 이 워커는 임대를 보유한 샤드당 레코드 프로세서 하나를 인스턴스화합니다.

checkpoint: 샤드의 가장 최근 체크포인트 시퀀스 번호입니다. 이 값은 데이터 스트림의 모든 샤드에서 고유합니다.

checkpointSubSequence번호: Kinesis 프로듀서 라이브러리의 집계 기능을 사용할 때 이는 Kinesis 레코드 내의 개별 사용자 레코드를 추적하는 체크포인트의 확장입니다.

leaseCounter: 임대 버전 관리에 사용되므로 작업자가 다른 근로자가 자신의 임대를 도용했음을 감지할 수 있습니다.

leaseKey: 임대의 고유 식별자. 각 리스는 데이터 스트림의 샤드에 특정적이며 한 번에 한 워커가 리스를 보유합니다.

leaseOwner: 이 임대차 계약을 체결한 근로자.

ownerSwitchesSince체크포인트: 마지막으로 체크포인트를 작성한 이후 이 임대 계약으로 인해 근로자가 몇 번이나 변경되었나요?

parentShardId: 하위 샤드에서 처리가 시작되기 전에 상위 샤드가 완전히 처리되었는지 확인하는 데 사용됩니다. 그러면 스트림에 입력된 순서대로 레코드가 처리됩니다.

hashrange: PeriodicShardSyncManager에서 주기적 동기화를 실행하여 리스 테이블에서 누락된 샤드를 찾고 필요한 경우 리스를 생성하는 데 사용됩니다.

childshards: LeaseCleanupManager에서 하위 샤드의 처리 상태를 검토하고 리스 테이블에서 상위 샤드를 삭제할 수 있는지 여부를 결정하는 데 사용됩니다.

shardID: 샤드의 ID입니다.

stream name account-id:StreamName:streamCreationTimestamp 형식의 데이터 스트림 식별자입니다.

처리 중인 데이터 스트림에 대한 모든 샤드를 가져옵니다.

리스 테이블에서 모든 샤드 리스를 가져옵니다.

리스 테이블에 리스가 없는 각 열린 샤드를 필터링합니다.

발견된 모든 열린 샤드와 열린 상위 항목이 없는 각 열린 샤드를 반복합니다.

애플리케이션을 부트스트래핑하는 동안 리스 테이블이 비어 있는 경우 ListShard API 의 필터링 옵션 (ShardFilter선택적 요청 파라미터) 을 KCL 사용하여 파라미터로 지정된 시간에 열린 샤드의 스냅샷에 대해서만 임대를 검색하고 생성합니다. ShardFilter ShardFilter파라미터를 사용하면 의 응답을 필터링할 수 있습니다. ListShards API ShardFilter 파라미터의 유일한 필수 속성은 Type입니다. KCLType필터 속성과 다음과 같은 유효한 값을 사용하여 새 임대가 필요할 수 있는 오픈 샤드의 스냅샷을 식별하고 반환합니다.

ShardFilter는 RetrievalConfig#initialPositionInStreamExtended에 지정된 샤드의 스냅샷에 대한 리스를 초기화하기 위해 빈 리스 테이블에 대한 리스를 생성할 때 사용됩니다.

ShardFilter에 대한 자세한 정보는 https://docs.aws.amazon.com/kinesis/latest/APIReference/API_ShardFilter.html을 참조하세요.

모든 워커가 리스/하드 동기화를 수행하여 데이터 스트림의 최신 샤드로 리스 테이블을 최신 상태로 유지하는 대신 선출된 단일 워커 리더가 리스/샤드 동기화를 수행합니다.

KCL2.3은 의 ChildShards 반환 매개 변수를 사용하여 폐쇄된 SubscribeToShard APIs 샤드의 SHARD_END 경우 발생하는 임대/샤드 동기화를 수행하므로 KCL 작업자는 처리가 완료된 샤드의 하위 샤드에 대해서만 임대를 생성할 수 있습니다. GetRecords 소비자 애플리케이션 전체에서 공유되는 경우 이 리스/샤드 동기화 최적화에서는 의 매개변수를 사용합니다. ChildShards GetRecords API 전용 처리량 (향상된 팬아웃) 소비자 애플리케이션의 경우 이러한 임대/샤드 동기화 최적화에는 의 매개 변수가 사용됩니다. ChildShards SubscribeToShard API 자세한 내용은, 및 을 참조하십시오. GetRecordsSubscribeToShardsChildShard

위의 변경으로 모든 작업자가 기존의 모든 샤드에 대해 학습하는 모델에서 작업자가 각 작업자가 소유한 샤드의 하위 샤드에 대해서만 학습하는 모델로 이동하고 있습니다. KCL 따라서 소비자 애플리케이션 부트스트래핑 및 리샤드 이벤트 중에 발생하는 동기화 외에도 KCL 이제는 추가 주기적인 샤드/리스 스캔을 수행하여 리스 테이블의 잠재적 허점을 식별 (즉, 모든 새 샤드에 대해 파악) 하여 데이터 스트림의 전체 해시 범위가 처리되고 있는지 확인하고 필요한 경우 리스를 생성합니다. PeriodicShardSyncManager주기적인 리스/샤드 스캔 실행을 담당하는 구성 요소입니다.

PeriodicShardSyncManagerIN KCL 2.3에 대한 자세한 내용은 /blob/master/ /src/main/java/software/amazon/kinesis/leases/ .java #L201 -L213을 참조하십시오 https://github.com/awslabs/amazon-kinesis-client. amazon-kinesis-client LeaseManagementConfig

2.3에서는 다음과 같은 새 구성 옵션을 구성할 수 있습니다. KCL PeriodicShardSyncManager LeaseManagementConfig

이제 의 상태를 모니터링하기 위한 새 CloudWatch 메트릭도 생성됩니다. PeriodicShardSyncManager 자세한 내용은 PeriodicShardSyncManager 단원을 참조하십시오.

하나의 샤드 계층에 대해서만 리스를 생성하도록 HierarchicalShardSyncer에 대한 최적화를 포함합니다.

애플리케이션을 부트스트래핑하는 동안 리스 테이블이 비어 있는 경우 ListShard API 의 필터링 옵션 (ShardFilter선택적 요청 파라미터) 을 KCL 사용하여 파라미터로 지정된 시간에 열린 샤드의 스냅샷에 대해서만 임대를 검색하고 생성합니다. ShardFilter ShardFilter파라미터를 사용하면 의 응답을 필터링할 수 있습니다. ListShards API ShardFilter 파라미터의 유일한 필수 속성은 Type입니다. KCLType필터 속성과 다음과 같은 유효한 값을 사용하여 새 임대가 필요할 수 있는 오픈 샤드의 스냅샷을 식별하고 반환합니다.

ShardFilter는 KinesisClientLibConfiguration#initialPositionInStreamExtended에 지정된 샤드의 스냅샷에 대한 리스를 초기화하기 위해 빈 리스 테이블에 대한 리스를 생성할 때 사용됩니다.

ShardFilter에 대한 자세한 정보는 https://docs.aws.amazon.com/kinesis/latest/APIReference/API_ShardFilter.html을 참조하세요.

모든 워커가 리스/하드 동기화를 수행하여 데이터 스트림의 최신 샤드로 리스 테이블을 최신 상태로 유지하는 대신 선출된 단일 워커 리더가 리스/샤드 동기화를 수행합니다.

KCL1.14는 의 ChildShards 반환 매개 변수를 사용하여 폐쇄된 SubscribeToShard APIs 샤드의 SHARD_END 경우 발생하는 임대/샤드 동기화를 수행하므로 KCL 작업자는 처리가 완료된 샤드의 하위 샤드에 대해서만 임대를 생성할 수 있습니다. GetRecords 자세한 내용은 및 GetRecordsChildShard을 참조하십시오.

위의 변경으로 모든 작업자가 기존의 모든 샤드에 대해 학습하는 모델에서 작업자가 각 작업자가 소유한 샤드의 하위 샤드에 대해서만 학습하는 모델로 이동하고 있습니다. KCL 따라서 소비자 애플리케이션 부트스트래핑 및 리샤드 이벤트 중에 발생하는 동기화 외에도 KCL 이제는 추가 주기적인 샤드/리스 스캔을 수행하여 리스 테이블의 잠재적 허점을 식별 (즉, 모든 새 샤드에 대해 파악) 하여 데이터 스트림의 전체 해시 범위가 처리되고 있는지 확인하고 필요한 경우 리스를 생성합니다. PeriodicShardSyncManager주기적인 리스/샤드 스캔 실행을 담당하는 구성 요소입니다.

KinesisClientLibConfiguration#shardSyncStrategyType이 ShardSyncStrategyType.SHARD_END로 설정된 경우 PeriodicShardSync leasesRecoveryAuditorInconsistencyConfidenceThreshold는 샤드 동기화 시행을 위한 리스 테이블에 구멍이 포함된 연속 스캔 수에 대한 임곗값을 결정하는 데 사용됩니다. KinesisClientLibConfiguration#shardSyncStrategyType이 ShardSyncStrategyType.PERIODIC으로 설정된 경우 leasesRecoveryAuditorInconsistencyConfidenceThreshold은 무시됩니다.

KCL1.14에 대한 자세한 내용은 PeriodicShardSyncManager /blob/v1.x/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/ .java #L987 -L999를 참조하십시오. https://github.com/awslabs/ amazon-kinesis-client KinesisClientLibConfiguration

1.14에서는 다음과 같은 새 구성 옵션을 사용하여 구성할 수 있습니다. KCL PeriodicShardSyncManager LeaseManagementConfig

이제 의 상태를 모니터링하기 위한 새 CloudWatch 메트릭도 생성됩니다. PeriodicShardSyncManager 자세한 내용은 PeriodicShardSyncManager 단원을 참조하십시오.

KCL1.14는 이제 지연 임대 정리도 지원합니다. 샤드가 데이터 스트림의 보존 기간을 지나 만료되었거나 리샤딩 작업의 결과로 닫힌 경우 SHARD_END에 도달하면 LeaseCleanupManager가 리스를 비동기식으로 삭제합니다.

새로운 구성 옵션을 사용하여 LeaseCleanupManager를 구성할 수 있습니다.

하나의 샤드 계층에 대해서만 리스를 생성하도록 KinesisShardSyncer에 대한 최적화를 포함합니다.

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

StreamIdentifier 및 InitialPositionInStreamExtended는 필수 필드이고 consumerArn은 선택 사항입니다. KCL2.x를 사용하여 향상된 팬아웃 소비자 애플리케이션을 구현하는 consumerArn 경우에만 를 제공해야 합니다.

에 대한 자세한 내용은 https://github.com/awslabs/amazon-kinesis-client/blob/v2.5.8/ StreamIdentifier /src/main/java/software/amazon/kinesis/common/ .java #L129 를 amazon-kinesis-client 참조하십시오. StreamIdentifier 를 생성하려면 v2.5.0 이상에서 사용할 수 있는 및 에서 멀티스트림 인스턴스를 생성하는 것이 좋습니다. StreamIdentifier streamArn streamCreationEpoch 지원하지 않는 KCL v2.3 및 v2.4에서는 다음 형식을 사용하여 streamArm 멀티스트림 인스턴스를 생성합니다. account-id:StreamName:streamCreationTimestamp 이 형식은 더 이상 사용되지 않으며 다음 주요 릴리스부터 더 이상 지원되지 않습니다.

MultistreamTracker에는 리스 테이블(formerStreamsLeasesDeletionStrategy)에서 오래된 스트림의 리스를 삭제하기 위한 전략도 포함되어 있습니다. 소비자 애플리케이션 런타임 중에 전략이 CANNOT 변경된다는 점에 유의하십시오. 자세한 내용은 https://github.com/awslabs/amazon-kinesis-client/blob/0c5042dadf794fe988438436252a5a8fe70b6b0b/ /src/main/java/software/amazon/kinesis/processor/.java를 amazon-kinesis-client 참조하십시오. FormerStreamsLeasesDeletionStrategy

ConfigsBuilder소비자 애플리케이션을 구축할 때 사용할 모든 2.x 구성 설정을 지정하는 데 사용할 수 있는 애플리케이션 전반의 클래스입니다. KCL KCL ConfigsBuilder클래스는 이제 인터페이스를 지원합니다. MultistreamTracker ConfigsBuilder둘 중 하나를 데이터 스트림의 이름으로 초기화하여 다음 데이터 스트림의 레코드를 소비할 수 있습니다.

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

또는 여러 스트림을 동시에 처리하는 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.