翻訳は機械翻訳により提供されています。提供された翻訳内容と英語版の間で齟齬、不一致または矛盾がある場合、英語版が優先します。

2KCL.x から 3.x KCL への移行

このトピックでは、 step-by-stepコンシューマーを 2.x から KCL 3.x KCL に移行する手順について説明します。 KCL 3.x は、2.x KCL コンシューマーのインプレース移行をサポートしています。ワーカーをローリング方式で移行しながら、Kinesis データストリームからのデータを引き続き消費できます。

ステップ 1: 前提条件

3.x KCL の使用を開始する前に、以下があることを確認してください。

ステップ 2: 依存関係を追加する

Maven を使用している場合は、次の依存関係を pom.xml ファイルに追加します。3.x.x をKCL最新バージョンに置き換えたことを確認してください。

<dependency>
    <groupId>software.amazon.kinesis</groupId>
    <artifactId>amazon-kinesis-client</artifactId>
    <version>3.x.x</version> <!-- Use the latest version -->
</dependency>

Gradle を使用している場合は、build.gradleファイルに以下を追加します。3.x.x をKCL最新バージョンに置き換えたことを確認してください。

implementation 'software.amazon.kinesis:amazon-kinesis-client:3.x.x'

Maven Central Repository KCLで の最新バージョンを確認できます。

ステップ 3: 移行関連の設定をセットアップする

2KCL.x から 3.x KCL に移行するには、次の設定パラメータを設定する必要があります。

configsBuilder.coordiantorConfig().clientVersionConfig(ClientVersionConfig.CLIENT_VERSION_CONFIG_COMPLATIBLE_WITH_2X)

2.x から 3KCL.x CoordinatorConfig.clientVersionConfig に移行するために を設定する方法の例を次に示します。必要に応じて、特定の要件に基づいて他の設定を調整できます。

Scheduler scheduler = new Scheduler(
    configsBuilder.checkpointConfig(),
    configsBuilder.coordiantorConfig().clientVersionConfig(ClientVersionConfig.CLIENT_VERSION_CONFIG_COMPLATIBLE_WITH_2X),
    configsBuilder.leaseManagementConfig(),
    configsBuilder.lifecycleConfig(),
    configsBuilder.metricsConfig(),
    configsBuilder.processorConfig(),
    configsBuilder.retrievalConfig()
);

KCL 2.x と 3.x では異なるロードバランシングアルゴリズムが使用されるため、コンシューマーアプリケーションのすべてのワーカーが同時に同じロードバランシングアルゴリズムを使用することが重要です。負荷分散アルゴリズムが異なるワーカーを実行すると、2 つのアルゴリズムが独立して動作するため、最適ではない負荷分散が発生する可能性があります。

この KCL 2.x 互換性設定により、コンシューマーアプリケーションのすべてのワーカーが KCL 3.x にアップグレードされるまで、3.x KCL アプリケーションは 2.x KCL と互換性のあるモードで実行され、2.x KCL の負荷分散アルゴリズムを使用できます。移行が完了すると、 KCLは自動的にフル KCL 3.x 機能モードに切り替え、実行中のすべてのワーカーに対して新しい KCL 3.x ロードバランシングアルゴリズムの使用を開始します。

ステップ 4: shutdownRequested() メソッド実装のベストプラクティスに従う

KCL 3.x では、リースの再割り当てプロセスの一環としてリースが別のワーカーに引き渡されたときのデータの再処理を最小限に抑えるために、グレースフルリースハンドオフと呼ばれる機能が導入されています。これは、リースの引き渡し前にリーステーブルで最後に処理されたシーケンス番号をチェックポイントすることで実現されます。リースの引き渡しが適切に機能するようにするには、 RecordProcessorクラスの shutdownRequestedメソッド内で checkpointer オブジェクトを呼び出す必要があります。shutdownRequested メソッド内でcheckpointerオブジェクトを呼び出しない場合は、次の例に示すように実装できます。

/**
 * Invoked when either Scheduler has been requested to gracefully shutdown
 * or lease ownership is being transferred gracefully so the current owner
 * gets one last chance to checkpoint.
 *
 * Checkpoints and logs the data a final time.
 *
 * @param shutdownRequestedInput Provides access to a checkpointer, allowing a record processor to checkpoint
 *                               before the shutdown is completed.
 */
public void shutdownRequested(ShutdownRequestedInput shutdownRequestedInput) {
    try {
       // Ensure that all delivered records are processed 
       // and has been successfully flushed to the downstream before calling 
       // checkpoint
       // If you are performing any asynchronous processing or flushing to
       // downstream, you must wait for its completion before invoking
       // the below checkpoint method.
        log.info("Scheduler is shutting down, checkpointing.");
        shutdownRequestedInput.checkpointer().checkpoint();
    } catch (ShutdownException | InvalidStateException e) {
        log.error("Exception while checkpointing at requested shutdown. Giving up.", e);
    } 
}

ステップ 5: ワーカーメトリクスを収集するための KCL 3.x の前提条件を確認する

KCL 3.x は、ワーカーのCPU使用率などのCPU使用率メトリクスを収集して、ワーカー間で負荷を均等に分散します。コンシューマーアプリケーションワーカーは、Amazon EC2、Amazon ECS、Amazon EKS、または で実行できます AWS Fargate。 KCL 3.x は、次の前提条件が満たされた場合にのみ、ワーカーからCPU使用率メトリクスを収集できます。

Amazon Elastic Compute Cloud(AmazonEC2）

Amazon での Amazon Elastic Container Service (Amazon ECS) EC2

ECSでの Amazon AWS Fargate

Amazon での Amazon Elastic Kubernetes Service (Amazon EKS) EC2

EKSでの Amazon AWS Fargate

ステップ 6: 3.x KCL のIAMアクセス許可を更新する

3KCL.x コンシューマーアプリケーションに関連付けられたIAMロールまたはポリシーに、次のアクセス許可を追加する必要があります。これには、KCLアプリケーションで使用される既存のIAMポリシーの更新が含まれます。詳細については、「IAM KCLコンシューマーアプリケーションに必要な アクセス許可」を参照してください。

ステップ 7: 3.x KCL コードをワーカーにデプロイする

移行に必要な設定を行い、以前の移行チェックリストをすべて完了したら、コードをビルドしてワーカーにデプロイできます。

ステップ 8: 移行を完了する

3.x KCL コードのデプロイ中、 は 2.x KCL のリース割り当てアルゴリズムをKCL引き続き使用します。すべてのワーカーに 3.x KCL コードを正常にデプロイすると、 KCLは自動的にこれを検出し、ワーカーのリソース使用率に基づいて新しいリース割り当てアルゴリズムに切り替えます。新しいリース割り当てアルゴリズムの詳細については、「」を参照してくださいがワーカーにリースをKCL割り当て、負荷のバランスをとる方法。

デプロイ中に、次のメトリクスを出力して移行プロセスをモニタリングできます CloudWatch。Migration オペレーションでメトリクスをモニタリングできます。すべてのメトリクスは per-KCL-applicationメトリクスであり、SUMMARYメトリクスレベルに設定されます。CurrentState:3xWorker メトリクスのSum統計がKCLアプリケーション内のワーカーの総数と一致する場合は、3.x KCL への移行が正常に完了したことを示します。

KCL は、移行中に 2.x 互換モードへのロールバック機能を提供します。3.x KCL への移行が成功したら、ロールバックが不要CLIENT_VERSION_CONFIG_COMPATIBLE_WITH_2Xになった場合は CoordinatorConfig.clientVersionConfigの設定を削除することをお勧めします。この設定を削除すると、KCLアプリケーションからの移行関連のメトリクスの放出が停止します。