KCL 1.x および 2.x の情報

KCL 1.x

詳細については、「1.x KCL コンシューマーを開発する」を参照してください

KCL 2.x

詳細については、「2.x KCL コンシューマーを開発する」を参照してください

KCL コンシューマーアプリケーション – を使用してカスタムビルドKCLされ、データストリームからレコードを読み取って処理するように設計されたアプリケーション。

コンシューマーアプリケーションインスタンス - KCLコンシューマーアプリケーションは通常、障害時に調整し、データレコード処理を動的に負荷分散するために、1 つ以上のアプリケーションインスタンスを同時に実行して分散されます。

ワーカー – KCLコンシューマーアプリケーションインスタンスがデータの処理を開始するために使用する高レベルクラス。

ワーカーは、シャードとリース情報の同期、シャード割り当ての追跡、シャードからのデータの処理など、さまざまなタスクを初期化し、監督します。ワーカーは、このコンシューマーアプリケーションKCLが処理するデータレコードのデータストリームの名前や、このデータストリームへのアクセスに必要な AWS 認証情報など、KCLコンシューマーアプリケーションの設定情報を提供します。また、ワーカーは、その特定のKCLコンシューマーアプリケーションインスタンスを起動して、データストリームからレコードプロセッサにデータレコードを配信します。

リース - ワーカーとシャード間のバインディングを定義するデータ。分散KCLコンシューマーアプリケーションは、リースを使用して、ワーカーのフリート間でデータレコード処理を分割します。いつでも、データレコードの各シャードは、 leaseKey変数によって識別されるリースによって特定のワーカーにバインドされます。

デフォルトでは、ワーカーは 1 つ以上のリースを同時に保持できます (maxLeasesForWorker 変数の値の影響を受けます）。

例えば、4 つのシャードを持つデータストリームを処理しているワーカー A を持つコンシューマーアプリケーションインスタンス A がある場合、ワーカー A はシャード 1、2、3、および 4 へのリースを同時に保持できます。ただし、2 つのコンシューマーアプリケーションインスタンス (ワーカー A とワーカー B を含む A と B) があり、これらのインスタンスが 4 つのシャードを持つデータストリームを処理している場合、ワーカー A とワーカー B はシャード 1 へのリースを同時に保持できません。あるワーカーは、このシャードのデータレコードの処理を停止する準備ができるまで、または失敗するまで、特定のシャードへのリースを保持します。あるワーカーがリースの保留を停止すると、別のワーカーがリースを引き取り、保留します。

詳細については、(Java 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 KCL の場合は https://github.com/awslabs/amazon-kinesis-client/blob/master/amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/Lease.javaKCL」を参照してください。

リーステーブル - KCLコンシューマーアプリケーションのワーカーによってリースおよび処理されるKDSデータストリーム内のシャードを追跡するために使用される一意の Amazon DynamoDB テーブル。リーステーブルは、KCLコンシューマーアプリケーションの実行中、データストリームからの最新のシャード情報と (ワーカー内およびすべてのワーカー間で) 同期している必要があります。詳細については、「リーステーブルを使用して、KCLコンシューマーアプリケーションによって処理されたシャードを追跡する」を参照してください。

レコードプロセッサ – KCLコンシューマーアプリケーションがデータストリームから取得したデータを処理する方法を定義するロジック。実行時に、KCLコンシューマーアプリケーションインスタンスはワーカーをインスタンス化し、このワーカーはリースを保持するシャードごとに 1 つのレコードプロセッサをインスタンス化します。

checkpoint: シャードの最新チェックポイントのシーケンス番号。この値はストリームのすべてのシャードで一意です。

checkpointSubSequence数値： Kinesis プロデューサーライブラリの集約機能を使用する場合、これは Kinesis レコード内の個々のユーザーレコードを追跡するチェックポイントの拡張機能です。

leaseCounter： リースのバージョニングに使用して、ワーカーが自分のリースが別のワーカーによって取得されたことを検出できるようにします。

leaseKey： リースの一意の識別子。各リースはデータストリームのシャードに固有であり、一度に 1 つのワーカーで保持されます。

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。 はType、フィルタープロパティとその有効な値の次の KCLを使用して、新しいリースを必要とする可能性のあるオープンシャードのスナップショットを識別して返します。

ShardFilter は空のリーステーブルのリースを作成して、RetrievalConfig#initialPositionInStreamExtended で指定したシャードのスナップショットのリースを初期化するときに使用されます。

ShardFilter の詳細については、「https://docs.aws.amazon.com/kinesis/latest/APIReference/API_ShardFilter.html」を参照してください。

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 同期を実行するすべてのワーカーの代わりに。

KCL 2.3 は、 GetRecordsと の ChildShards return パラメータSubscribeToShardAPIsを使用して、閉じたシャードSHARD_ENDに対して で発生するリース/シャード同期を実行し、KCLワーカーが処理が終了したシャードの子シャードのリースのみを作成できるようにします。コンシューマーアプリケーション全体で共有する場合、リース/シャード同期のこの最適化ではGetRecords、 の ChildShardsパラメータを使用しますAPI。専用スループット (拡張ファンアウト) コンシューマーアプリケーションの場合、リース/シャード同期のこの最適化ではSubscribeToShard、 の ChildShardsパラメータを使用しますAPI。詳細については、GetRecords、SubscribeToShards、ChildShard を参照してください。

上記の変更により、 の動作KCLは、既存のすべてのシャードについて学習するすべてのワーカーのモデルから、各ワーカーが所有するシャードの子シャードについてのみ学習するワーカーのモデルに移行します。したがって、コンシューマーアプリケーションのブートストラップおよびリシャードイベント中に発生する同期に加えて、 KCL は追加の定期的なシャード/リーススキャンも実行し、リーステーブルの潜在的な穴 (つまり、すべての新しいシャードについて学習する) を特定して、データストリームの完全なハッシュ範囲が処理されていることを確認し、必要に応じてリースを作成します。 PeriodicShardSyncManagerは、定期的なリース/シャードスキャンの実行を担当するコンポーネントです。

2.3 KCL PeriodicShardSyncManagerの の詳細については、https://github.com/awslabs/amazon-kinesis-client「/blob/master/amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/LeaseManagementConfig.java#L201-L213」を参照してください。

KCL 2.3 では、新しい設定オプションが PeriodicShardSyncManagerで設定できるようになりましたLeaseManagementConfig。

の正常性をモニタリングするために、新しい CloudWatch メトリクスも出力されるようになりましたPeriodicShardSyncManager。詳細については、「PeriodicShardSyncManager」を参照してください。

HierarchicalShardSyncer への最適化を含めて、シャードの 1 つのレイヤーに対してのみリースを作成します。

アプリケーションのブートストラップ中に、リーステーブルが空の場合、 は ListShard APIのフィルタリングオプション (ShardFilterオプションのリクエストパラメータ) KCLを使用して、 ShardFilterパラメータで指定された時間に開かれたシャードのスナップショットのみを取得してリースを作成します。ShardFilter パラメータを使用すると、 ListShards のレスポンスを除外できますAPI。ShardFilter パラメータの唯一の必須プロパティは ですType。 KCLはType、フィルタープロパティとその有効な値のうち、新しいリースを必要とする可能性のあるオープンシャードのスナップショットを識別して返します。

ShardFilter は空のリーステーブルのリースを作成して、KinesisClientLibConfiguration#initialPositionInStreamExtended で指定したシャードのスナップショットのリースを初期化するときに使用されます。

ShardFilter の詳細については、「https://docs.aws.amazon.com/kinesis/latest/APIReference/API_ShardFilter.html」を参照してください。

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 同期を実行するすべてのワーカーの代わりに。

KCL 1.14 は、 GetRecordsおよび の ChildShards return パラメータSubscribeToShardAPIsを使用して、閉じたシャードSHARD_ENDに対して で発生するリース/シャード同期を実行し、KCLワーカーが処理が終了したシャードの子シャードのリースのみを作成できるようにします。詳細については、GetRecordsおよびChildShardを参照してください。

上記の変更により、 の動作KCLは、既存のすべてのシャードについて学習するすべてのワーカーのモデルから、各ワーカーが所有するシャードの子シャードについてのみ学習するワーカーのモデルに移行します。したがって、コンシューマーアプリケーションのブートストラップおよびリシャードイベント中に発生する同期に加えて、 KCL は追加の定期的なシャード/リーススキャンも実行し、リーステーブルの潜在的な穴 (つまり、すべての新しいシャードについて学習する) を特定して、データストリームの完全なハッシュ範囲が処理されていることを確認し、必要に応じてリースを作成します。 PeriodicShardSyncManagerは、定期的なリース/シャードスキャンの実行を担当するコンポーネントです。

KinesisClientLibConfiguration#shardSyncStrategyType が ShardSyncStrategyType.SHARD_END に設定されると、PeriodicShardSync leasesRecoveryAuditorInconsistencyConfidenceThreshold は、シャード同期を強制するために、リーステーブル内のホールを含む連続スキャンの数のしきい値を決定するために使用されます。KinesisClientLibConfiguration#shardSyncStrategyType が ShardSyncStrategyType.PERIODIC に設定されると、leasesRecoveryAuditorInconsistencyConfidenceThreshold は無視されます。

1.14 PeriodicShardSyncManagerの KCL の詳細については、https://github.com/awslabs/amazon-kinesis-client「/blob/v1.x/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/KinesisClientLibConfiguration.java#L987-L999」を参照してください。

KCL 1.14 では、新しい設定オプションが PeriodicShardSyncManagerで設定できるようになりましたLeaseManagementConfig。

の正常性をモニタリングするために、新しい CloudWatch メトリクスも出力されるようになりましたPeriodicShardSyncManager。詳細については、「PeriodicShardSyncManager」を参照してください。

KCL 1.14 では、遅延リースのクリーンアップもサポートされるようになりました。リースは、SHARD_END に到達したとき、シャードがデータストリームの保持期間を過ぎて期限切れになったとき、またはリシャーディングオペレーションの結果として閉じられたとき、LeaseCleanupManager により非同期的に削除されます。

新しい設定オプションを使用して、LeaseCleanupManager を設定できるようになりました。

KinesisShardSyncer への最適化を含めて、シャードの 1 つのレイヤーに対してのみリースを作成します。

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 は省略可能である点に注意してください。2.x KCL を使用して拡張ファンアウトコンシューマーアプリケーションを実装しているconsumerArn場合にのみ、 を指定する必要があります。

の詳細についてはStreamIdentifier、https://github.com/awslabs/amazon-kinesis-client「/blob/v2.5.8/amazon-kinesis-client/src/main/java/software/amazon/kinesis/common/StreamIdentifier.java#L129」を参照してください。StreamIdentifier を作成するには、streamArn および v2.5.0 以降で利用可能な streamCreationEpoch からマルチストリームインスタンスを作成することをお勧めします。をサポートしていない KCL v2.3 および v2.4 ではstreamArm、 形式を使用してマルチストリームインスタンスを作成しますaccount-id:StreamName:streamCreationTimestamp。この形式は廃止され、次のメジャーリリース以降はサポートされなくなります。

MultistreamTracker には、リーステーブル内の古いストリームのリースを削除するための戦略も含まれます(formerStreamsLeasesDeletionStrategy)。コンシューマーアプリケーションのランタイム中に戦略CANNOTが変更されることに注意してください。詳細については、https://github.com/awslabs/「amazon-kinesis-client/blob/0c5042dadf794fe988438436252a5a8fe70b6b0b/amazon-kinesis-client.java」を参照してくださいsrc/main/java/software/amazon/kinesis/processor/FormerStreamsLeasesDeletionStrategy。

ConfigsBuilder は、KCLコンシューマーアプリケーションの構築時に使用するすべての KCL 2.x 構成設定を指定するために使用できるアプリケーション全体のクラスです。 ConfigsBuilder クラスで MultistreamTrackerインターフェイスがサポートされるようになりました。レコードを消費する 1 つのデータストリームの名前を使用して、次のいずれかを初期化 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 と指定します。