翻訳は機械翻訳により提供されています。提供された翻訳内容と英語版の間で齟齬、不一致または矛盾がある場合、英語版が優先します。
# 共有スループットでカスタムコンシューマーを開発する
**重要**
Amazon Kinesis Client Library (KCL) バージョン 1.x および 2.x は古くなっています。KCL 1.x は 2026 年 1 月 30 日にサポートが終了します。2026 年 1 月 30 日より前に、バージョン 1.x を使用して KCL アプリケーションを最新の KCL バージョンに移行することを**強くお勧めします**。最新の KCL バージョンを確認するには、[GitHub のAmazon Kinesis Client Library](https://github.com/awslabs/amazon-kinesis-client)ページを参照してください。最新の KCL バージョンの詳細については、[Kinesis Client Library を使用する](kcl.md) を参照してください。KCL 1.x から KCL 3.x への移行については、「[KCL 1.x から KCL 3.x への移行](kcl-migration-1-3.md)」を参照してください。
Kinesis Data Streams からデータを受け取る際に専用スループットを必要としない場合で、200 ms 以下の読み取り伝達遅延を必要としない場合は、以下のトピックで説明しているようにコンシューマーアプリケーションを構築できます。Kinesis Client Library (KCL) または AWS SDK for Javaを使用できます。
**Topics**
+ [KCL を使用した共有スループットでカスタムコンシューマーを開発する](custom-kcl-consumers.md)
専有スループットで Kinesis data streams からレコードを受信できるコンシューマーの構築の詳細については、[専用スループットを備えた拡張ファンアウトを開発する](enhanced-consumers.md)を参照してください。
# KCL を使用した共有スループットでカスタムコンシューマーを開発する
**重要**
Amazon Kinesis Client Library (KCL) バージョン 1.x および 2.x は古くなっています。KCL 1.x は 2026 年 1 月 30 日にサポートが終了します。2026 年 1 月 30 日より前に、バージョン 1.x を使用して KCL アプリケーションを最新の KCL バージョンに移行することを**強くお勧めします**。最新の KCL バージョンを確認するには、[GitHub のAmazon Kinesis Client Library](https://github.com/awslabs/amazon-kinesis-client)ページを参照してください。最新の KCL バージョンの詳細については、[Kinesis Client Library を使用する](kcl.md) を参照してください。KCL 1.x から KCL 3.x への移行については、「[KCL 1.x から KCL 3.x への移行](kcl-migration-1-3.md)」を参照してください。
共有スループットでカスタムコンシューマーアプリケーションを開発する方法の 1 つは、Kinesis Client Library (KCL) を使用することです。
使用している KCL バージョンの次のトピックから選択します。
**Topics**
+ [KCL 1.x コンシューマーを開発する](developing-consumers-with-kcl.md)
+ [KCL 2.x コンシューマーを開発する](developing-consumers-with-kcl-v2.md)
# KCL 1.x コンシューマーを開発する
**重要**
Amazon Kinesis Client Library (KCL) バージョン 1.x および 2.x は古くなっています。KCL 1.x は 2026 年 1 月 30 日にサポートが終了します。2026 年 1 月 30 日より前に、バージョン 1.x を使用して KCL アプリケーションを最新の KCL バージョンに移行することを**強くお勧めします**。最新の KCL バージョンを確認するには、[GitHub のAmazon Kinesis Client Library](https://github.com/awslabs/amazon-kinesis-client)ページを参照してください。最新の KCL バージョンの詳細については、[Kinesis Client Library を使用する](kcl.md) を参照してください。KCL 1.x から KCL 3.x への移行については、「[KCL 1.x から KCL 3.x への移行](kcl-migration-1-3.md)」を参照してください。
Kinesis Client Library (KCL) を使用して、Amazon Kinesis Data Streams のコンシューマーアプリケーションを開発することができます。
KCL の詳細については、[KCL について (以前のバージョン)](shared-throughput-kcl-consumers.md#shared-throughput-kcl-consumers-overview)を参照してください。
使用するオプションに応じて、次のトピックから選択します。
**Topics**
+ [Java での Kinesis クライアントライブラリコンシューマーを開発する](kinesis-record-processor-implementation-app-java.md)
+ [ode.js で Kinesis Client Library コンシューマーを開発する](kinesis-record-processor-implementation-app-nodejs.md)
+ [.NET で Kinesis Client Library コンシューマーを開発する](kinesis-record-processor-implementation-app-dotnet.md)
+ [Python で Kinesis クライアントライブラリコンシューマーを開発する](kinesis-record-processor-implementation-app-py.md)
+ [Ruby で Kinesis Client Library コンシューマーを開発する](kinesis-record-processor-implementation-app-ruby.md)
# Java での Kinesis クライアントライブラリコンシューマーを開発する
**重要**
Amazon Kinesis Client Library (KCL) バージョン 1.x および 2.x は古くなっています。KCL 1.x は 2026 年 1 月 30 日にサポートが終了します。2026 年 1 月 30 日より前に、バージョン 1.x を使用して KCL アプリケーションを最新の KCL バージョンに移行することを**強くお勧めします**。最新の KCL バージョンを確認するには、[GitHub のAmazon Kinesis Client Library](https://github.com/awslabs/amazon-kinesis-client)ページを参照してください。最新の KCL バージョンの詳細については、[Kinesis Client Library を使用する](kcl.md) を参照してください。KCL 1.x から KCL 3.x への移行については、「[KCL 1.x から KCL 3.x への移行](kcl-migration-1-3.md)」を参照してください。
Kinesis Data Streams のデータを処理するアプリケーションを構築するには Kinesis Client Library (KCL) を使用できます。Kinesis Client Library は、複数の言語で使用できます。このトピックでは、Java について説明します。Javadoc リファレンスを表示するには、[AWS Javadoc topic for Class AmazonKinesisClient](https://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/com/amazonaws/services/kinesis/AmazonKinesisClient.html)を参照してください。
GitHub から Java KCL をダウンロードするには、[Kinesis Client Library (Java)](https://github.com/awslabs/amazon-kinesis-client) にアクセスしてください。Apache Maven で Java KCL を検索するには、[KCL 検索結果](https://search.maven.org/#search|ga|1|amazon-kinesis-client)のページを参照してください。Java KCL コンシューマーアプリケーションのサンプルコードをダウンロードするには、GitHub の[KCL for Java sample project](https://github.com/aws/aws-sdk-java/tree/master/src/samples/AmazonKinesis)ページを参照してください。
このサンプルアプリケーションは [Apache Commons Logging](http://commons.apache.org/proper/commons-logging/guide.html) を使用します。ログ設定は、`configure` ファイルで定義されている静的な `AmazonKinesisApplicationSample.java` メソッドを使用して変更できます。Log4j および Java アプリケーションで Apache Commons ログ記録を使用する方法の詳細については、「 *AWS SDK for Java デベロッパーガイド*」の「Log4j AWS を使用したログ記録」を参照してください。 [ Log4j](https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/java-dg-logging.html)
Java で KCL コンシューマーアプリケーションを実装する場合は、次のタスクを完了する必要があります。
**Topics**
+ [IRecordProcessor メソッドを実装する](#kinesis-record-processor-implementation-interface-java)
+ [IRecordProcessor インターフェイスのクラスファクトリを実装する](#kinesis-record-processor-implementation-factory-java)
+ [ワーカーを作成する](#kcl-java-worker)
+ [設定プロパティを変更する](#kinesis-record-processor-initialization-java)
+ [レコードプロセッサインターフェイスのバージョン 2 に移行する](#kcl-java-v2-migration)
## IRecordProcessor メソッドを実装する
KCL は現在、`IRecordProcessor` インターフェイスの 2 つのバージョンをサポートしています。元のインターフェイスは最初のバージョンの KCL で利用可能です。バージョン 2 は KCL バージョン 1.5.0 から利用可能です。両方のインターフェイスが完全にサポートされています。選択するインターフェイスは、お使いのシナリオの要件によって異なります。相違点をすべて確認するには、ローカルに作成した Javadocs、またはソースコードを参照してください。以下のセクションでは、使い始めの最小限の実装を概説します。
**Topics**
+ [オリジナルインターフェイス (バージョン 1)](#kcl-java-interface-original)
+ [更新されたインターフェイス (バージョン 2)](#kcl-java-interface-v2)
### オリジナルインターフェイス (バージョン 1)
オリジナルな `IRecordProcessor` interface (`package com.amazonaws.services.kinesis.clientlibrary.interfaces`) は、コンシューマーが実装しているべき次のレコードプロセッサメソッドを公開します。このサンプルでは、開始点として使用できる実装を提供しています (`AmazonKinesisApplicationSampleRecordProcessor.java` を参照してください)。
```
public void initialize(String shardId)
public void processRecords(List records, IRecordProcessorCheckpointer checkpointer)
public void shutdown(IRecordProcessorCheckpointer checkpointer, ShutdownReason reason)
```
**初期化**
KCL は、レコードプロセッサがインスタンス化されると、`initialize` メソッドを呼び出し、特定のシャード ID をパラメータとして渡します。このレコードプロセッサはこのシャードのみを処理し、通常、その逆も真です (このシャードはこのレコード プロセッサによってのみ処理されます)。ただし、コンシューマーでは、データレコードが複数回処理される可能性に対応する必要があります。Kinesis Data Streams は*少なくとも 1 回*のセマンティクスを使用しています。これは、シャードから取得されたすべてのデータレコードが、コンシューマーのワーカーによって少なくとも 1 回処理されることを意味します。特定のシャードが複数のワーカーによって処理される可能性がある場合の詳細については、[シャードの数を変更するには、再シャーディング、スケーリング、並列処理を使用します。](kinesis-record-processor-scaling.md)を参照してください。
```
public void initialize(String shardId)
```
**processRecords**
KCL は、`processRecords` メソッドを呼び出し、`initialize(shardId)` メソッドで指定されたシャードのデータレコードのリストを渡します。レコードプロセッサは、コンシューマーのセマンティクスに従って、これらのレコードのデータを処理します。例えば、ワーカーはデータの変換を実行し、その結果を Amazon Simple Storage Service (Amazon S3) バケットに保存する場合があります。
```
public void processRecords(List records, IRecordProcessorCheckpointer checkpointer)
```
データ自体に加えて、レコードにもシーケンス番号とパーティションキーが含まれます。ワーカーはデータを処理するときに、これらの値を使用できます。たとえば、ワーカーは、パーティションのキーの値に基づいて、データを格納する S3 バケットを選択できます。`Record` クラスは、レコードのデータ、シーケンス番号、およびパーティションキーへのアクセスを提供する次のメソッドを公開します。
```
record.getData()
record.getSequenceNumber()
record.getPartitionKey()
```
サンプルでは、プライベートメソッド `processRecordsWithRetries` に、ワーカーでレコードのデータ、シーケンス番号、およびパーティションキーにアクセスする方法を示すコードが含まれています。
Kinesis Data Streams では、シャードで既に処理されたレコードを追跡するためにレコードプロセッサが必要です。KCL は、チェックポインタ (`IRecordProcessorCheckpointer`) を `processRecords` に渡すことで、この追跡をユーザーに代わって処理します。レコードプロセッサは、このインターフェイスで `checkpoint` メソッドを呼び出し、シャード内のレコードの処理の進行状況を KCL に知らせます。ワーカーでエラーが発生すると、KCL はこの情報を使用して、処理されたことが分かっている最後のレコードからシャードの処理を再開します。
分割または結合オペレーションの場合、KCL は、元のシャードのプロセッサが `checkpoint` を呼び出して元のシャードの処理がすべて完了したことを通知するまで、新しいシャードの処理を開始しません。
パラメータを渡さないと、`checkpoint` への呼び出しは、レコードプロセッサに最後のレコードを渡した時点までのすべてのレコードが処理済みであることを意味すると KCL で見なされます。したがって、レコードプロセッサは、渡されたリストにあるすべてのレコードの処理が完了した場合にのみ、`checkpoint` を呼び出す必要があります。レコードプロセッサは、`checkpoint` の各呼び出しで `processRecords` を呼び出す必要はありません。たとえば、プロセッサは、`checkpoint` を 3 回呼び出すたびに、`processRecords` を呼び出すことができます。オプションでレコードの正確なシーケンス番号をパラメータとして `checkpoint` に指定できます。この場合、KCL は、すべてのレコードがそのレコードまで処理されたと見なします。
このサンプルでは、プライベートメソッド `checkpoint` で、適切な例外処理と再試行のロジックを使用する `IRecordProcessorCheckpointer.checkpoint` を呼び出す方法を示しています。
KCL は、`processRecords` を使用して、データレコードの処理から発生するすべての例外を処理します。例外が `processRecords` からスローされた場合、KCL は、例外発生前に渡されたデータレコードをスキップします。つまり、これらのレコードは、例外をスローしたレコードプロセッサ、またはコンシューマーの他のレコードプロセッサに再送信されません。
**シャットダウン**
KCL は、処理が終了した場合 (シャットダウンの理由は `TERMINATE`) またはワーカーが応答していない場合 (シャットダウンの理由は `ZOMBIE`)、`shutdown` メソッドを呼び出します。
```
public void shutdown(IRecordProcessorCheckpointer checkpointer, ShutdownReason reason)
```
シャードが分割または結合されたか、ストリームが削除されたため、レコードプロセッサがシャードからこれ以上レコードを受信しない場合は、処理が終了します。
KCL はまた、`IRecordProcessorCheckpointer` インターフェイスを `shutdown` に渡します。シャットダウンの理由が `TERMINATE` である場合、レコードプロセッサはすべてのデータレコードの処理を終了し、このインターフェイスの `checkpoint` メソッドを呼び出します。
### 更新されたインターフェイス (バージョン 2)
更新された `IRecordProcessor` interface (`package com.amazonaws.services.kinesis.clientlibrary.interfaces.v2`) は、コンシューマーが実装しているべき次のレコードプロセッサメソッドを公開します。
```
void initialize(InitializationInput initializationInput)
void processRecords(ProcessRecordsInput processRecordsInput)
void shutdown(ShutdownInput shutdownInput)
```
コンテナオブジェクトのメソッドの呼び出しで、インターフェイスのオリジナルバージョンのすべての引数にアクセスできます。たとえば、`processRecords()` でレコードのリストを取得には、`processRecordsInput.getRecords()` が使用できます。
このインターフェイスのバージョン 2 (KCL 1.5.0 以降) では、オリジナルインターフェースで提供される入力に加えて次の新しい入力が使用できます。
シーケンス番号の開始
`InitializationInput` オペレーションへ渡される `initialize()` オブジェクトでは、開始シーケンス番号はレコードプロセッサのインスタンスに配信されるレコードです。このシーケンス番号は、同じシャードで処理されたレコードプロセッサインスタンスの最後のチェックポイントです。これは、アプリケーションでこの情報が必要になる場合のために提供されます。
保留チェックポイントシーケンス番号
`initialize()` オペレーションへ渡される `InitializationInput`オブジェクトの保留チェックポイントシーケンス番号 (ある場合) とは、前のレコードプロセッサインスタンスが停止する前にコミットできなかったものを示します。
## IRecordProcessor インターフェイスのクラスファクトリを実装する
レコードプロセッサのメソッドを実装するクラスのファクトリも実装する必要があります。コンシューマーは、ワーカーをインスタンス化するときに、このファクトリへの参照を渡します。
サンプルでは、オリジナルのレコードプロセッサインターフェースを使用した、`AmazonKinesisApplicationSampleRecordProcessorFactory.java` ファイルのファクトリクラスを実装します。クラスファクトリでバージョン 2 レコードプロセッサを作成する場合には、`com.amazonaws.services.kinesis.clientlibrary.interfaces.v2` とい名のパッケージを使用してください。
```
public class SampleRecordProcessorFactory implements IRecordProcessorFactory {
/**
* Constructor.
*/
public SampleRecordProcessorFactory() {
super();
}
/**
* {@inheritDoc}
*/
@Override
public IRecordProcessor createProcessor() {
return new SampleRecordProcessor();
}
}
```
## ワーカーを作成する
[IRecordProcessor メソッドを実装する](#kinesis-record-processor-implementation-interface-java)で説明しているように、KCL レコードプロセッサには選択できる 2 バージョンがあり、どちらを選ぶかでワーカーの作成方法に影響します。オリジナルレコードプロセッサインターフェイスは、次のコードストラクチャを使用してワーカーを作成します。
```
final KinesisClientLibConfiguration config = new KinesisClientLibConfiguration(...)
final IRecordProcessorFactory recordProcessorFactory = new RecordProcessorFactory();
final Worker worker = new Worker(recordProcessorFactory, config);
```
レコード プロセッサインターフェイスのバージョン 2 では、`Worker.Builder` を使用してワーカを作成でき、どのコンストラクタを使うかや引数の順序を考慮する必要はありません。更新されたレコードプロセッサインターフェイスは、次のコードストラクチャを使用してワーカーを作成します。
```
final KinesisClientLibConfiguration config = new KinesisClientLibConfiguration(...)
final IRecordProcessorFactory recordProcessorFactory = new RecordProcessorFactory();
final Worker worker = new Worker.Builder()
.recordProcessorFactory(recordProcessorFactory)
.config(config)
.build();
```
## 設定プロパティを変更する
このサンプルでは、設定プロパティのデフォルト値を提供します。ワーカーのこの設定データは `KinesisClientLibConfiguration` オブジェクトにまとめられています。ワーカーをインスタンス化する呼び出しで、このオブジェクトと `IRecordProcessor` のクラスファクトリへの参照が渡されます。Java の properties ファイルを使用してこれらのプロパティを独自の値にオーバーライドできます (`AmazonKinesisApplicationSample.java` を参照してください)。
### アプリケーション名
KCL には、複数のアプリケーション間、および同じリージョン内の Amazon DynamoDB テーブル間で一意のアプリケーション名が必要です。次のようにアプリケーション名の設定値を使用します。
+ このアプリケーション名と関連付けられたすべてのワーカーは、連係して同じストリームを処理していると見なされます。これらのワーカーは複数のインスタンスに分散している場合もあります。同じアプリケーションコードの追加のインスタンスを実行するときに、アプリケーション名が異なる場合、KCL は 2 番目のインスタンスを、同じストリームで動作するまったく別のアプリケーションと見なします。
+ KCL はアプリケーション名を使用して DynamoDB テーブルを作成し、このテーブルを使用してアプリケーションの状態情報 (チェックポイントやワーカーとシャードのマッピングなど) を保存します。各アプリケーションには、それぞれ DynamoDB テーブルがあります。詳細については、[リーステーブルを使用して KCL コンシューマーアプリケーションによって処理されたシャードを追跡する](shared-throughput-kcl-consumers.md#shared-throughput-kcl-consumers-leasetable)を参照してください。
### 認証情報の設定
デフォルトの AWS 認証情報プロバイダーチェーンの認証情報プロバイダーのいずれかが認証情報を利用できるようにする必要があります。例えば、EC2 インスタンスでコンシューマーを実行している場合は、IAM ロールでインスタンスを起動することをお勧めします。この IAM ロールに関連付けられた許可を反映する AWS 認証情報は、インスタンスメタデータを通じて、インスタンス上のアプリケーションで使用できるようになります。これは、EC2 インスタンスで実行されるコンシューマーの認証情報を管理するための最も安全な方法です。
サンプルアプリケーションは、最初にインスタンスメタデータから IAM 認証情報を取得しようとします。
```
credentialsProvider = new InstanceProfileCredentialsProvider();
```
サンプルアプリケーションは、インスタンスメタデータから認証情報を取得できない場合、properties ファイルから認証情報を取得しようとします。
```
credentialsProvider = new ClasspathPropertiesFileCredentialsProvider();
```
インスタンスメタデータの詳細については、「Amazon EC2 ユーザーガイド」の「[インスタンスメタデータ](https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/ec2-instance-metadata.html)」を参照してください。**
### 複数のインスタンスへのワーカー ID を使用する
サンプルの初期化コードは、次のコードスニペットに示すように、ローカルコンピュータ名にグローバル一意識別子を追加して、ワーカーの ID (`workerId`) を作成します。このアプローチによって、1 台のコンピュータでコンシューマーアプリケーションの複数のインスタンスを実行するシナリオに対応できます。
```
String workerId = InetAddress.getLocalHost().getCanonicalHostName() + ":" + UUID.randomUUID();
```
## レコードプロセッサインターフェイスのバージョン 2 に移行する
オリジナルインターフェースで使われるコードを移行するためには、上記のステップに加えて、次の手順が必要となります。
1. レコードプロセッサのクラスを変更して、バージョン 2 レコードプロセッサインターフェイスにインポートします。
```
import com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IRecordProcessor;
```
1. コンテナオブジェクトで `get` メソッドを使用するには、入力するリファレンスを変更します。たとえば、`shutdown()` オペレーションで、`checkpointer` を `shutdownInput.getCheckpointer()` に変更します。
1. レコードプロセッサのファクトリークラスを変更して、バージョン 2 レコードプロセッサファクトリーインターフェイスにインポートします。
```
import com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IRecordProcessorFactory;
```
1. ワーカーのコンストラクチャを変更して、`Worker.Builder` を使います。例えば、次のようになります。
```
final Worker worker = new Worker.Builder()
.recordProcessorFactory(recordProcessorFactory)
.config(config)
.build();
```
# ode.js で Kinesis Client Library コンシューマーを開発する
**重要**
Amazon Kinesis Client Library (KCL) バージョン 1.x および 2.x は古くなっています。KCL 1.x は 2026 年 1 月 30 日にサポートが終了します。2026 年 1 月 30 日より前に、バージョン 1.x を使用して KCL アプリケーションを最新の KCL バージョンに移行することを**強くお勧めします**。最新の KCL バージョンを確認するには、[GitHub のAmazon Kinesis Client Library](https://github.com/awslabs/amazon-kinesis-client)ページを参照してください。最新の KCL バージョンの詳細については、[Kinesis Client Library を使用する](kcl.md) を参照してください。KCL 1.x から KCL 3.x への移行については、「[KCL 1.x から KCL 3.x への移行](kcl-migration-1-3.md)」を参照してください。
Kinesis Data Streams のデータを処理するアプリケーションを構築するには Kinesis Client Library (KCL) を使用できます。Kinesis Client Library は、複数の言語で使用できます。このトピックでは、Node.js について説明します。
KCL は Java ライブラリであり、Java 以外の言語のサポートは、*MultiLangDaemon* と呼ばれる多言語インターフェースを使用して提供されます。このデーモンは Java ベースで、Java 以外の KCL 言語を使用しているときにバックグラウンドで実行されます。そのため、KCL for Node.js をインストールして、コンシューマーアプリケーションをすべて Node.js で書く場合でも、MultiLangDaemon を使用するために、Java をシステムにインストールする必要があります。さらに、MultiLangDaemon には、接続先の AWS リージョンなど、ユースケースに合わせてカスタマイズする必要があるデフォルト設定があります。GitHub の MultiLangDaemon の詳細については、[KCL MultiLangDaemon project](https://github.com/awslabs/amazon-kinesis-client/tree/v1.x/src/main/java/com/amazonaws/services/kinesis/multilang)のページを参照してください。
GitHub から Java KCL をダウンロードするには、[Kinesis Client Library (Node.js)](https://github.com/awslabs/amazon-kinesis-client-nodejs) にアクセスしてください。
**サンプルコードのダウンロード**
Node.js の KCL で使用可能な 2 つのサンプルコードがあります。
+ [基本サンプル](https://github.com/awslabs/amazon-kinesis-client-nodejs/tree/master/samples/basic_sample)
Node.js で KCL コンシューマーアプリケーションを構築する方法の基本を説明する次のセクションで使用されます。
+ [click-stream-sample](https://github.com/awslabs/amazon-kinesis-client-nodejs/tree/master/samples/click_stream_sample)
基本サンプルコードを理解したあとの、やや上級で実際のシナリオを使用したサンプル。このサンプルはここでは説明しませんが、詳細を説明した README ファイルがあります。
Node.js で KCL コンシューマーアプリケーションを実装する場合は、次のタスクを完了する必要があります。
**Topics**
+ [レコードプロセッサを実装する](#kinesis-record-processor-implementation-interface-nodejs)
+ [設定プロパティを変更する](#kinesis-record-processor-initialization-nodejs)
## レコードプロセッサを実装する
KCL for Node.js を使用した最もシンプルなコンシューマーは、`recordProcessor` 関数を実装する必要があります。この関数には、`initialize`、`processRecords`、および `shutdown` の各関数が含まれます。このサンプルでは、開始点として使用できる実装を提供しています (`sample_kcl_app.js` を参照してください)。
```
function recordProcessor() {
// return an object that implements initialize, processRecords and shutdown functions.}
```
**初期化**
レコードプロセッサが起動すると、KCL は `initialize` 関数を呼び出します。このレコードプロセッサは `initializeInput.shardId` として渡されるシャード ID のみを処理し、通常、その逆も真です (このシャードはこのレコードプロセッサによってのみ処理されます)。ただし、コンシューマーでは、データレコードが複数回処理される可能性に対応する必要があります。これは、Kinesis Data Streams は*少なくとも 1 回*のセマンティクスを使用しているからです。つまり、シャードから取得されたすべてのデータレコードが、コンシューマーのワーカーによって少なくとも 1 回処理されることを意味します。特定のシャードが複数のワーカーによって処理される可能性がある場合の詳細については、[シャードの数を変更するには、再シャーディング、スケーリング、並列処理を使用します。](kinesis-record-processor-scaling.md)を参照してください。
```
initialize: function(initializeInput, completeCallback)
```
**processRecords**
KCL は、この関数を呼び出すために `initialize` 関数に指定したシャードのデータレコードのリストが含まれている入力を使用します。実装するレコードプロセッサは、コンシューマーのセマンティクスに従って、これらのレコードのデータを処理します。例えば、ワーカーはデータの変換を実行し、その結果を Amazon Simple Storage Service (Amazon S3) バケットに保存する場合があります。
```
processRecords: function(processRecordsInput, completeCallback)
```
データ自体に加えて、レコードにもシーケンス番号とパーティションキーが含まれ、ワーカーはデータを処理するときに、これらを使用できます。たとえば、ワーカーは、パーティションのキーの値に基づいて、データを格納する S3 バケットを選択できます。`record` ディクショナリは、レコードのデータ、シーケンス番号、およびパーティションキーにアクセスする次のキーと値のペアを公開します。
```
record.data
record.sequenceNumber
record.partitionKey
```
データは Base64 でエンコードされていることに注意してください。
基本サンプルでは、関数 `processRecords` に、ワーカーでレコードのデータ、シーケンス番号、およびパーティションキーにアクセスする方法を示すコードが含まれています。
Kinesis Data Streams では、シャードで既に処理されたレコードを追跡するためにレコードプロセッサが必要です。KCL は、`processRecordsInput.checkpointer` として渡した `checkpointer` オブジェクトを使用して、この追跡を処理します。レコードプロセッサは、`checkpointer.checkpoint` 関数を呼び出して、シャード内のレコードの処理の進行状況を KCL に知らせます。ワーカーでエラーが発生した場合、シャードの処理を再開するときに、処理されたことが分かっている最後のレコードから再開するように、KCL はこの情報を使用します。
分割または結合オペレーションの場合、KCL は、元のシャードのプロセッサが `checkpoint` を呼び出して元のシャードの処理がすべて完了したことを通知するまで、新しいシャードの処理を開始しません。
`checkpoint` 関数にシーケンス番号を渡さないと、`checkpoint` への呼び出しは、レコードプロセッサに最後のレコードを渡した時点までのすべてのレコードが処理済みであることを意味すると KCL で見なされます。したがって、レコードプロセッサは、渡されたリストにあるすべてのレコードの処理が完了した場合に**のみ**、`checkpoint` を呼び出す必要があります。レコードプロセッサは、`checkpoint` の各呼び出しで `processRecords` を呼び出す必要はありません。たとえば、プロセッサは `checkpoint` を 3 回の呼び出しごとに呼び出したり、レコードプロセッサの外部イベント (実装したカスタムの認証または検証サービスなど) で呼び出したりできます。
オプションでレコードの正確なシーケンス番号をパラメータとして `checkpoint` に指定できます。この場合、KCL は、そのレコードまでのすべてのレコードだけが処理されたと見なします。
基本サンプルアプリケーションでは、`checkpointer.checkpoint` 関数の最もシンプルな呼び出しを示します。関数のこの時点でコンシューマーに必要な他のチェックポイントロジックを追加できます。
**シャットダウン**
KCL は、処理が終了した場合 (`shutdownInput.reason` は `TERMINATE`) またはワーカーが応答していない場合 (`shutdownInput.reason` は `ZOMBIE`)、`shutdown` 関数を呼び出します。
```
shutdown: function(shutdownInput, completeCallback)
```
シャードが分割または結合されたか、ストリームが削除されたため、レコードプロセッサがシャードからこれ以上レコードを受信しない場合は、処理が終了します。
また、KCL は、`shutdownInput.checkpointer` オブジェクトも `shutdown` に渡します。シャットダウンの理由が `TERMINATE` である場合、レコードプロセッサがすべてのデータレコードの処理を終了したことを確認し、このインターフェイスの `checkpoint` 関数を呼び出します。
## 設定プロパティを変更する
このサンプルでは、設定プロパティのデフォルト値を提供します。これらのプロパティを独自の値にオーバーライドできます (基本サンプルの `sample.properties` を参照してください)。
### アプリケーション名
KCL には、複数のアプリケーション間、および同じリージョン内の Amazon DynamoDB テーブル間で一意のアプリケーションが必要です。次のようにアプリケーション名の設定値を使用します。
+ このアプリケーション名と関連付けられたすべてのワーカーは、連係して同じストリームを処理していると見なされます。これらのワーカーは複数のインスタンスに分散している場合もあります。同じアプリケーションコードの追加のインスタンスを実行するときに、アプリケーション名が異なる場合、KCL は 2 番目のインスタンスを、同じストリームで動作するまったく別のアプリケーションと見なします。
+ KCL はアプリケーション名を使用して DynamoDB テーブルを作成し、このテーブルを使用してアプリケーションの状態情報 (チェックポイントやワーカーとシャードのマッピングなど) を保存します。各アプリケーションには、それぞれ DynamoDB テーブルがあります。詳細については、[リーステーブルを使用して KCL コンシューマーアプリケーションによって処理されたシャードを追跡する](shared-throughput-kcl-consumers.md#shared-throughput-kcl-consumers-leasetable)を参照してください。
### 認証情報の設定
デフォルトの AWS 認証情報プロバイダーチェーンの認証情報プロバイダーのいずれかが認証情報を利用できるようにする必要があります。`AWSCredentialsProvider` プロパティを使用して認証情報プロバイダーを設定できます。`sample.properties` ファイルでは、[デフォルトの認証情報プロバイダーチェーン](https://docs.aws.amazon.com/sdk-for-java/latest/reference/com/amazonaws/auth/DefaultAWSCredentialsProviderChain.html)のいずれかの認証情報プロバイダーに対して、ユーザーの認証情報を使用可能にする必要があります。Amazon EC2 インスタンスでコンシューマーを実行している場合は、この IAM ロールに関連付けられたアクセス許可を反映する IAM role. AWS credentials を使用してインスタンスを設定することをお勧めします。この IAM ロールは、インスタンスメタデータを介してインスタンス上のアプリケーションで使用できます。これは、EC2 インスタンスで実行されるコンシューマーアプリケーションの認証情報を管理するための最も安全な方法です。
次の例では、KCL を設定し、`sample_kcl_app.js` で指定されているレコードプロセッサを使用して`kclnodejssample`という Kinesis Data Streams を処理します。
```
# The Node.js executable script
executableName = node sample_kcl_app.js
# The name of an Amazon Kinesis stream to process
streamName = kclnodejssample
# Unique KCL application name
applicationName = kclnodejssample
# Use default AWS credentials provider chain
AWSCredentialsProvider = DefaultAWSCredentialsProviderChain
# Read from the beginning of the stream
initialPositionInStream = TRIM_HORIZON
```
# .NET で Kinesis Client Library コンシューマーを開発する
**重要**
Amazon Kinesis Client Library (KCL) バージョン 1.x および 2.x は古くなっています。KCL 1.x は 2026 年 1 月 30 日にサポートが終了します。2026 年 1 月 30 日より前に、バージョン 1.x を使用して KCL アプリケーションを最新の KCL バージョンに移行することを**強くお勧めします**。最新の KCL バージョンを確認するには、[GitHub のAmazon Kinesis Client Library](https://github.com/awslabs/amazon-kinesis-client)ページを参照してください。最新の KCL バージョンの詳細については、[Kinesis Client Library を使用する](kcl.md) を参照してください。KCL 1.x から KCL 3.x への移行については、「[KCL 1.x から KCL 3.x への移行](kcl-migration-1-3.md)」を参照してください。
Kinesis Data Streams のデータを処理するアプリケーションを構築するには Kinesis Client Library (KCL) を使用できます。Kinesis Client Library は、複数の言語で使用できます。このトピックでは、.NET について説明します。
KCL は Java ライブラリであり、Java 以外の言語のサポートは、*MultiLangDaemon* と呼ばれる多言語インターフェースを使用して提供されます。このデーモンは Java ベースで、Java 以外の KCL 言語を使用しているときにバックグラウンドで実行されます。そのため、KCL for .NET をインストールして、コンシューマーアプリケーションをすべて .NET で書く場合でも、MultiLangDaemon を使用するために、Java をシステムにインストールする必要があります。さらに、MultiLangDaemon には、接続先の AWS リージョンなど、ユースケースに合わせてカスタマイズする必要があるデフォルト設定があります。GitHub の MultiLangDaemon の詳細については、[KCL MultiLangDaemon project](https://github.com/awslabs/amazon-kinesis-client/tree/v1.x/src/main/java/com/amazonaws/services/kinesis/multilang)のページを参照してください。
GitHub から .NET KCL をダウンロードするには、[Kinesis Client Library (.NET)](https://github.com/awslabs/amazon-kinesis-client-net) にアクセスしてください。.NET KCL コンシューマーアプリケーションのサンプルコードをダウンロードするには、GitHub で[KCL for .NET sample consumer project](https://github.com/awslabs/amazon-kinesis-client-net/tree/master/SampleConsumer) のページにアクセスしてください。
.NET で KCL コンシューマーアプリケーションを実装する場合は、次のタスクを完了する必要があります。
**Topics**
+ [IRecordProcessor クラスのメソッドを実装する](#kinesis-record-processor-implementation-interface-dotnet)
+ [設定プロパティを変更する](#kinesis-record-processor-initialization-dotnet)
## IRecordProcessor クラスのメソッドを実装する
コンシューマーでは、`IRecordProcessor` の次のメソッドを実装する必要があります。出発点として使用できる実装がサンプルコンシューマーに提供されています (`SampleRecordProcessor` の `SampleConsumer/AmazonKinesisSampleConsumer.cs` クラスを参照してください)。
```
public void Initialize(InitializationInput input)
public void ProcessRecords(ProcessRecordsInput input)
public void Shutdown(ShutdownInput input)
```
**Initialize**
KCL は、レコードプロセッサがインスタンス化されると、このメソッドを呼び出して `input` パラメータの特定のシャード ID (`input.ShardId`) を渡します。このレコードプロセッサはこのシャードのみを処理し、通常、その逆も真です (このシャードはこのレコード プロセッサによってのみ処理されます)。ただし、コンシューマーでは、データレコードが複数回処理される可能性に対応する必要があります。これは、Kinesis Data Streams は*少なくとも 1 回*のセマンティクスを使用しているからです。つまり、シャードから取得されたすべてのデータレコードが、コンシューマーのワーカーによって少なくとも 1 回処理されることを意味します。特定のシャードが複数のワーカーによって処理される可能性がある場合の詳細については、[シャードの数を変更するには、再シャーディング、スケーリング、並列処理を使用します。](kinesis-record-processor-scaling.md)を参照してください。
```
public void Initialize(InitializationInput input)
```
**ProcessRecords**
KCL は、このメソッドを呼び出し、`Initialize` メソッドで指定されたシャードの `input` パラメータ (`input.Records`) にあるデータレコードのリストを渡します。実装するレコードプロセッサは、コンシューマーのセマンティクスに従って、これらのレコードのデータを処理します。例えば、ワーカーはデータの変換を実行し、その結果を Amazon Simple Storage Service (Amazon S3) バケットに保存する場合があります。
```
public void ProcessRecords(ProcessRecordsInput input)
```
データ自体に加えて、レコードにもシーケンス番号とパーティションキーが含まれます。ワーカーはデータを処理するときに、これらの値を使用できます。たとえば、ワーカーは、パーティションのキーの値に基づいて、データを格納する S3 バケットを選択できます。`Record` クラスは以下を公開し、レコードのデータ、シーケンス番号、およびパーティションキーのアクセスを可能にします。
```
byte[] Record.Data
string Record.SequenceNumber
string Record.PartitionKey
```
サンプルでは、メソッド `ProcessRecordsWithRetries` に、ワーカーでレコードのデータ、シーケンス番号、およびパーティションキーにアクセスする方法を示すコードが含まれています。
Kinesis Data Streams では、シャードで既に処理されたレコードを追跡するためにレコードプロセッサが必要です。KCL は、`Checkpointer` オブジェクトを `ProcessRecords` に渡すことで、この追跡をユーザーに代わって処理します (`input.Checkpointer`)。レコードプロセッサは、`Checkpointer.Checkpoint` メソッドを呼び出して、シャード内のレコード処理の進行状況を KCL に知らせます。ワーカーでエラーが発生すると、KCL はこの情報を使用して、処理されたことが分かっている最後のレコードからシャードの処理を再開します。
分割または結合オペレーションの場合、KCL は、元のシャードのプロセッサが `Checkpointer.Checkpoint` を呼び出して元のシャードの処理がすべて完了したことを通知するまで、新しいシャードの処理を開始しません。
パラメータを渡さないと、`Checkpointer.Checkpoint` への呼び出しは、レコードプロセッサに最後のレコードを渡した時点までのすべてのレコードが処理済みであることを意味すると KCL で見なされます。したがって、レコードプロセッサは、渡されたリストにあるすべてのレコードの処理が完了した場合にのみ、`Checkpointer.Checkpoint` を呼び出す必要があります。レコードプロセッサは、`Checkpointer.Checkpoint` の各呼び出しで `ProcessRecords` を呼び出す必要はありません。たとえば、プロセッサは、3 回または 4 回呼び出すたびに、`Checkpointer.Checkpoint` を呼び出すことができます。オプションでレコードの正確なシーケンス番号をパラメータとして `Checkpointer.Checkpoint` に指定できます。この場合、KCL は、レコード処理がそのレコードまで完了したと見なします。
サンプルでは、プライベートメソッド `Checkpoint(Checkpointer checkpointer)` で、適切な例外処理と再試行のロジックを使用する `Checkpointer.Checkpoint` メソッドを呼び出す方法を示しています。
KCL for .NET では、例外を処理する方法が他の KCL 言語ライブラリとは異なり、データレコードの処理から発生した例外を扱いません。ユーザーコードからの例外がキャッチされないと、プログラムがクラッシュします。
**シャットダウン**
KCL は、処理が終了した場合 (シャットダウンの理由は `TERMINATE`) またはワーカーが応答していない場合 (シャットダウンの `input.Reason` の値は `ZOMBIE`)、`Shutdown` メソッドを呼び出します。
```
public void Shutdown(ShutdownInput input)
```
シャードが分割または結合されたか、ストリームが削除されたため、レコードプロセッサがシャードからこれ以上レコードを受信しない場合は、処理が終了します。
また、KCL は、`Checkpointer` オブジェクトも `shutdown` に渡します。シャットダウンの理由が `TERMINATE` である場合、レコードプロセッサはすべてのデータレコードの処理を終了し、このインターフェイスの `checkpoint` メソッドを呼び出します。
## 設定プロパティを変更する
このサンプルコンシューマーでは、設定プロパティのデフォルト値を提供します。これらのプロパティを独自の値にオーバーライドできます (`SampleConsumer/kcl.properties` を参照してください)。
### アプリケーション名
KCL には、複数のアプリケーション間、および同じリージョン内の Amazon DynamoDB テーブル間で一意のアプリケーションが必要です。次のようにアプリケーション名の設定値を使用します。
+ このアプリケーション名と関連付けられたすべてのワーカーは、連係して同じストリームを処理していると見なされます。これらのワーカーは複数のインスタンスに分散している場合もあります。同じアプリケーションコードの追加のインスタンスを実行するときに、アプリケーション名が異なる場合、KCL は 2 番目のインスタンスを、同じストリームで動作するまったく別のアプリケーションと見なします。
+ KCL はアプリケーション名を使用して DynamoDB テーブルを作成し、このテーブルを使用してアプリケーションの状態情報 (チェックポイントやワーカーとシャードのマッピングなど) を保存します。各アプリケーションには、それぞれ DynamoDB テーブルがあります。詳細については、[リーステーブルを使用して KCL コンシューマーアプリケーションによって処理されたシャードを追跡する](shared-throughput-kcl-consumers.md#shared-throughput-kcl-consumers-leasetable)を参照してください。
### 認証情報の設定
デフォルトの AWS 認証情報プロバイダーチェーンの認証情報プロバイダーのいずれかが認証情報を利用できるようにする必要があります。`AWSCredentialsProvider` プロパティを使用して認証情報プロバイダーを設定できます。[sample.properties](https://github.com/awslabs/amazon-kinesis-client-python/blob/master/samples/sample.properties) では、[デフォルトの認証情報プロバイダーチェーン](https://docs.aws.amazon.com/sdk-for-java/latest/reference/com/amazonaws/auth/DefaultAWSCredentialsProviderChain.html)のいずれかの認証情報プロバイダーに対して、ユーザーの認証情報を使用可能にする必要があります。EC2 インスタンスでコンシューマーアプリケーションを実行している場合は、IAM ロールでインスタンスを設定することをお勧めします。この IAM ロールに関連付けられた許可を反映する AWS 認証情報は、インスタンスメタデータを通じて、インスタンス上のアプリケーションで使用できるようになります。これは、EC2 インスタンスで実行されるコンシューマーの認証情報を管理するための最も安全な方法です。
サンプルのプロパティファイルでは、 で指定されているレコードプロセッサを使用してwordsという Kinesis data stream を処理するように KCL を設定します。
# Python で Kinesis クライアントライブラリコンシューマーを開発する
**重要**
Amazon Kinesis Client Library (KCL) バージョン 1.x および 2.x は古くなっています。KCL 1.x は 2026 年 1 月 30 日にサポートが終了します。2026 年 1 月 30 日より前に、バージョン 1.x を使用して KCL アプリケーションを最新の KCL バージョンに移行することを**強くお勧めします**。最新の KCL バージョンを確認するには、[GitHub のAmazon Kinesis Client Library](https://github.com/awslabs/amazon-kinesis-client)ページを参照してください。最新の KCL バージョンの詳細については、[Kinesis Client Library を使用する](kcl.md) を参照してください。KCL 1.x から KCL 3.x への移行については、「[KCL 1.x から KCL 3.x への移行](kcl-migration-1-3.md)」を参照してください。
Kinesis Data Streams のデータを処理するアプリケーションを構築するには Kinesis Client Library (KCL) を使用できます。Kinesis Client Library は、複数の言語で使用できます。このトピックでは、Python について説明します。
KCL は Java ライブラリであり、Java 以外の言語のサポートは、*MultiLangDaemon* と呼ばれる多言語インターフェースを使用して提供されます。このデーモンは Java ベースで、Java 以外の KCL 言語を使用しているときにバックグラウンドで実行されます。そのため、KCL for Python をインストールして、コンシューマーアプリケーションをすべて Python で書く場合でも、MultiLangDaemon を使用するために、Java をシステムにインストールする必要があります。さらに、MultiLangDaemon には、接続先の AWS リージョンなど、ユースケースに合わせてカスタマイズする必要があるデフォルト設定があります。GitHub の MultiLangDaemon の詳細については、[KCL MultiLangDaemon project](https://github.com/awslabs/amazon-kinesis-client/tree/v1.x/src/main/java/com/amazonaws/services/kinesis/multilang)のページを参照してください。
GitHub から Python KCL をダウンロードするには、[Kinesis Client Library (Python)](https://github.com/awslabs/amazon-kinesis-client-python) にアクセスしてください。Python KCL コンシューマーアプリケーションのサンプルコードをダウンロードするには、GitHub で[KCL for Python sample project](https://github.com/awslabs/amazon-kinesis-client-python/tree/master/samples)ページにアクセスしてください。
Python で KCL コンシューマーアプリケーションを実装する場合は、次のタスクを完了する必要があります。
**Topics**
+ [RecordProcessor クラスのメソッドを実装する](#kinesis-record-processor-implementation-interface-py)
+ [設定プロパティを変更する](#kinesis-record-processor-initialization-py)
## RecordProcessor クラスのメソッドを実装する
`RecordProcess` クラスでは、`RecordProcessorBase` を拡張して次のメソッドを実装する必要があります。このサンプルでは、開始点として使用できる実装を提供しています (`sample_kclpy_app.py` を参照してください)。
```
def initialize(self, shard_id)
def process_records(self, records, checkpointer)
def shutdown(self, checkpointer, reason)
```
**初期化**
KCL は、レコードプロセッサがインスタンス化されると、`initialize` メソッドを呼び出し、特定のシャード ID をパラメータとして渡します。このレコードプロセッサはこのシャードのみを処理し、通常、その逆も真です (このシャードはこのレコード プロセッサによってのみ処理されます)。ただし、コンシューマーでは、データレコードが複数回処理される可能性に対応する必要があります。これは、Kinesis Data Streams は*少なくとも 1 回*のセマンティクスを使用しているからです。つまり、シャードから取得されたすべてのデータレコードが、コンシューマーのワーカーによって少なくとも 1 回処理されることを意味します。特定のシャードが複数のワーカーによって処理される可能性がある場合の詳細については、[シャードの数を変更するには、再シャーディング、スケーリング、並列処理を使用します。](kinesis-record-processor-scaling.md)を参照してください。
```
def initialize(self, shard_id)
```
**process\$1records**
KCL は、このメソッドを呼び出し、`initialize` メソッドで指定されたシャードのデータレコードのリストを渡します。実装するレコードプロセッサは、コンシューマーのセマンティクスに従って、これらのレコードのデータを処理します。例えば、ワーカーはデータの変換を実行し、その結果を Amazon Simple Storage Service (Amazon S3) バケットに保存する場合があります。
```
def process_records(self, records, checkpointer)
```
データ自体に加えて、レコードにもシーケンス番号とパーティションキーが含まれます。ワーカーはデータを処理するときに、これらの値を使用できます。たとえば、ワーカーは、パーティションのキーの値に基づいて、データを格納する S3 バケットを選択できます。`record` ディクショナリは、レコードのデータ、シーケンス番号、およびパーティションキーにアクセスする次のキーと値のペアを公開します。
```
record.get('data')
record.get('sequenceNumber')
record.get('partitionKey')
```
データは Base64 でエンコードされていることに注意してください。
サンプルでは、メソッド `process_records` に、ワーカーでレコードのデータ、シーケンス番号、およびパーティションキーにアクセスする方法を示すコードが含まれています。
Kinesis Data Streams では、シャードで既に処理されたレコードを追跡するためにレコードプロセッサが必要です。KCL は、`Checkpointer` オブジェクトを `process_records` に渡すことで、この追跡をユーザーに代わって処理します。レコードプロセッサは、このオブジェクトの `checkpoint` メソッドを呼び出して、シャード内のレコードの処理の進行状況を KCL に通知します。ワーカーでエラーが発生すると、KCL はこの情報を使用して、処理されたことが分かっている最後のレコードからシャードの処理を再開します。
分割または結合オペレーションの場合、KCL は、元のシャードのプロセッサが `checkpoint` を呼び出して元のシャードの処理がすべて完了したことを通知するまで、新しいシャードの処理を開始しません。
パラメータを渡さないと、`checkpoint` への呼び出しは、レコードプロセッサに最後のレコードを渡した時点までのすべてのレコードが処理済みであることを意味すると KCL で見なされます。したがって、レコードプロセッサは、渡されたリストにあるすべてのレコードの処理が完了した場合にのみ、`checkpoint` を呼び出す必要があります。レコードプロセッサは、`checkpoint` の各呼び出しで `process_records` を呼び出す必要はありません。たとえば、プロセッサは、3 回呼び出すたびに、`checkpoint` を呼び出すことができます。オプションでレコードの正確なシーケンス番号をパラメータとして `checkpoint` に指定できます。この場合、KCL は、そのレコードまでのすべてのレコードだけが処理されたと見なします。
サンプルでは、プライベートメソッド `checkpoint` で、適切な例外処理と再試行のロジックを使用する `Checkpointer.checkpoint` メソッドを呼び出す方法を示しています。
KCL は、`process_records` を使用して、データレコードの処理から発生するすべての例外を処理します。例外が `process_records` からスローされた場合、KCL は、例外発生前に `process_records` に渡されたデータレコードをスキップします。つまり、これらのレコードは、例外をスローしたレコードプロセッサ、またはコンシューマーの他のレコードプロセッサに再送信されません。
**シャットダウン**
KCL は、処理が終了した場合 (シャットダウンの理由は `TERMINATE`) またはワーカーが応答していない場合 (シャットダウンの `reason` は `ZOMBIE`)、`shutdown` メソッドを呼び出します。
```
def shutdown(self, checkpointer, reason)
```
シャードが分割または結合されたか、ストリームが削除されたため、レコードプロセッサがシャードからこれ以上レコードを受信しない場合は、処理が終了します。
また、KCL は、`Checkpointer` オブジェクトも `shutdown` に渡します。シャットダウンの `reason` が `TERMINATE` である場合、レコードプロセッサはすべてのデータレコードの処理を終了し、このインターフェイスの `checkpoint` メソッドを呼び出します。
## 設定プロパティを変更する
このサンプルでは、設定プロパティのデフォルト値を提供します。これらのプロパティを独自の値にオーバーライドできます (`sample.properties` を参照してください)。
### アプリケーション名
KCL には、複数のアプリケーション間、および同じリージョン内の Amazon DynamoDB テーブル間で一意のアプリケーションが必要です。次のようにアプリケーション名の設定値を使用します。
+ このアプリケーション名と関連付けられたワーカーはすべて、同じストリーム上で連携して処理しているとみなされます。これらのワーカーは複数のインスタンスに分散している場合があります。同じアプリケーションコードの追加のインスタンスを実行するときに、アプリケーション名が異なる場合、KCL は 2 番目のインスタンスを、同じストリームで動作するまったく別のアプリケーションと見なします。
+ KCL はアプリケーション名を使用して DynamoDB テーブルを作成し、このテーブルを使用してアプリケーションの状態情報 (チェックポイントやワーカーとシャードのマッピングなど) を保存します。各アプリケーションには、それぞれ DynamoDB テーブルがあります。詳細については、[リーステーブルを使用して KCL コンシューマーアプリケーションによって処理されたシャードを追跡する](shared-throughput-kcl-consumers.md#shared-throughput-kcl-consumers-leasetable)を参照してください。
### 認証情報の設定
デフォルトの AWS 認証情報プロバイダーチェーンの認証情報プロバイダーのいずれかが認証情報を利用できるようにする必要があります。`AWSCredentialsProvider` プロパティを使用して認証情報プロバイダーを設定できます。[sample.properties](https://github.com/awslabs/amazon-kinesis-client-python/blob/master/samples/sample.properties) では、[デフォルトの認証情報プロバイダーチェーン](https://docs.aws.amazon.com/sdk-for-java/latest/reference/com/amazonaws/auth/DefaultAWSCredentialsProviderChain.html)のいずれかの認証情報プロバイダーに対して、ユーザーの認証情報を使用可能にする必要があります。Amazon EC2 インスタンスでコンシューマーアプリケーションを実行している場合は、IAM ロールでインスタンスを設定することをお勧めします。この IAM ロールに関連付けられた許可を反映する AWS 認証情報は、インスタンスメタデータを通じて、インスタンス上のアプリケーションで使用できるようになります。これは、EC2 インスタンスで実行されるコンシューマーアプリケーションの認証情報を管理するための最も安全な方法です。
サンプルのプロパティファイルでは、 で指定されているレコードプロセッサを使用してwordsという Kinesis data stream を処理するように KCL を設定します。
# Ruby で Kinesis Client Library コンシューマーを開発する
**重要**
Amazon Kinesis Client Library (KCL) バージョン 1.x および 2.x は古くなっています。KCL 1.x は 2026 年 1 月 30 日にサポートが終了します。2026 年 1 月 30 日より前に、バージョン 1.x を使用して KCL アプリケーションを最新の KCL バージョンに移行することを**強くお勧めします**。最新の KCL バージョンを確認するには、[GitHub のAmazon Kinesis Client Library](https://github.com/awslabs/amazon-kinesis-client)ページを参照してください。最新の KCL バージョンの詳細については、[Kinesis Client Library を使用する](kcl.md) を参照してください。KCL 1.x から KCL 3.x への移行については、「[KCL 1.x から KCL 3.x への移行](kcl-migration-1-3.md)」を参照してください。
Kinesis Data Streams のデータを処理するアプリケーションを構築するには Kinesis Client Library (KCL) を使用できます。Kinesis Client Library は、複数の言語で使用できます。このトピックでは、Ruby について説明します。
KCL は Java ライブラリであり、Java 以外の言語のサポートは、*MultiLangDaemon* と呼ばれる多言語インターフェースを使用して提供されます。このデーモンは Java ベースで、Java 以外の KCL 言語を使用しているときにバックグラウンドで実行されます。そのため、KCL for .Ruby をインストールして、コンシューマーアプリケーションをすべて Ruby で書く場合でも、MultiLangDaemon を使用するために、Java をシステムにインストールする必要があります。さらに、MultiLangDaemon には、接続先の AWS リージョンなど、ユースケースに合わせてカスタマイズする必要があるデフォルト設定があります。GitHub の MultiLangDaemon の詳細については、[KCL MultiLangDaemon project](https://github.com/awslabs/amazon-kinesis-client/tree/v1.x/src/main/java/com/amazonaws/services/kinesis/multilang)のページを参照してください。
GitHub から Ruby KCL をダウンロードするには、[Kinesis Client Library (Ruby)](https://github.com/awslabs/amazon-kinesis-client-ruby) にアクセスしてください。Ruby KCL コンシューマーアプリケーションのサンプルコードをダウンロードするには、GitHub で[KCL for Ruby sample project](https://github.com/awslabs/amazon-kinesis-client-ruby/tree/master/samples)ページにアクセスしてください。
KCL Ruby サポートライブラリの詳細については、[KCL Ruby Gems ドキュメント](http://www.rubydoc.info/gems/aws-kclrb)を参照してください。
# KCL 2.x コンシューマーを開発する
**重要**
Amazon Kinesis Client Library (KCL) バージョン 1.x および 2.x は古くなっています。KCL 1.x は 2026 年 1 月 30 日にサポートが終了します。2026 年 1 月 30 日より前に、バージョン 1.x を使用して KCL アプリケーションを最新の KCL バージョンに移行することを**強くお勧めします**。最新の KCL バージョンを確認するには、[GitHub のAmazon Kinesis Client Library](https://github.com/awslabs/amazon-kinesis-client)ページを参照してください。最新の KCL バージョンの詳細については、[Kinesis Client Library を使用する](kcl.md) を参照してください。KCL 1.x から KCL 3.x への移行については、「[KCL 1.x から KCL 3.x への移行](kcl-migration-1-3.md)」を参照してください。
このトピックでは、バージョン 2.0 の Kinesis Client Library (KCL) を使用する方法について説明します。
KCL の詳細については、[Kinesis Client Library 1.x を使用したコンシューマーの開発](https://docs.aws.amazon.com/streams/latest/dev/developing-consumers-with-kcl.html)に示されている概要を参照してください。
使用するオプションに応じて、次のトピックから選択します。
**Topics**
+ [Java での Kinesis クライアントライブラリコンシューマーを開発する](kcl2-standard-consumer-java-example.md)
+ [Python で Kinesis クライアントライブラリコンシューマーを開発する](kcl2-standard-consumer-python-example.md)
+ [KCL 2.x を使用して拡張ファンアウトコンシューマーを開発する](building-enhanced-consumers-kcl-retired.md)
# Java での Kinesis クライアントライブラリコンシューマーを開発する
**重要**
Amazon Kinesis Client Library (KCL) バージョン 1.x および 2.x は古くなっています。KCL 1.x は 2026 年 1 月 30 日にサポートが終了します。2026 年 1 月 30 日より前に、バージョン 1.x を使用して KCL アプリケーションを最新の KCL バージョンに移行することを**強くお勧めします**。最新の KCL バージョンを確認するには、[GitHub のAmazon Kinesis Client Library](https://github.com/awslabs/amazon-kinesis-client)ページを参照してください。最新の KCL バージョンの詳細については、[Kinesis Client Library を使用する](kcl.md) を参照してください。KCL 1.x から KCL 3.x への移行については、「[KCL 1.x から KCL 3.x への移行](kcl-migration-1-3.md)」を参照してください。
次のコードは、`ProcessorFactory` および `RecordProcessor` の Java のサンプル実装を示しています。拡張ファンアウト機能を活用する方法については、[拡張ファンアウトでコンシューマーを使用する](https://docs.aws.amazon.com/streams/latest/dev/building-enhanced-consumers-kcl-java.html)を参照してください。
```
/*
* Copyright 2019 Amazon.com, Inc. or its affiliates. All Rights Reserved.
*
* Licensed under the Amazon Software License (the "License").
* You may not use this file except in compliance with the License.
* A copy of the License is located at
*
* http://aws.amazon.com/asl/
*
* or in the "license" file accompanying this file. This file is distributed
* on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
* express or implied. See the License for the specific language governing
* permissions and limitations under the License.
*/
/*
* Copyright 2019 Amazon.com, Inc. or its affiliates. All Rights Reserved.
*
* Licensed under the Apache License, Version 2.0 (the "License").
* You may not use this file except in compliance with the License.
* A copy of the License is located at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* or in the "license" file accompanying this file. This file is distributed
* on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
* express or implied. See the License for the specific language governing
* permissions and limitations under the License.
*/
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.UUID;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.apache.commons.lang3.ObjectUtils;
import org.apache.commons.lang3.RandomStringUtils;
import org.apache.commons.lang3.RandomUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;
import software.amazon.awssdk.core.SdkBytes;
import software.amazon.awssdk.regions.Region;
import software.amazon.awssdk.services.cloudwatch.CloudWatchAsyncClient;
import software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClient;
import software.amazon.awssdk.services.kinesis.KinesisAsyncClient;
import software.amazon.awssdk.services.kinesis.model.PutRecordRequest;
import software.amazon.kinesis.common.ConfigsBuilder;
import software.amazon.kinesis.common.KinesisClientUtil;
import software.amazon.kinesis.coordinator.Scheduler;
import software.amazon.kinesis.exceptions.InvalidStateException;
import software.amazon.kinesis.exceptions.ShutdownException;
import software.amazon.kinesis.lifecycle.events.InitializationInput;
import software.amazon.kinesis.lifecycle.events.LeaseLostInput;
import software.amazon.kinesis.lifecycle.events.ProcessRecordsInput;
import software.amazon.kinesis.lifecycle.events.ShardEndedInput;
import software.amazon.kinesis.lifecycle.events.ShutdownRequestedInput;
import software.amazon.kinesis.processor.ShardRecordProcessor;
import software.amazon.kinesis.processor.ShardRecordProcessorFactory;
import software.amazon.kinesis.retrieval.polling.PollingConfig;
/**
* This class will run a simple app that uses the KCL to read data and uses the AWS SDK to publish data.
* Before running this program you must first create a Kinesis stream through the AWS console or AWS SDK.
*/
public class SampleSingle {
private static final Logger log = LoggerFactory.getLogger(SampleSingle.class);
/**
* Invoke the main method with 2 args: the stream name and (optionally) the region.
* Verifies valid inputs and then starts running the app.
*/
public static void main(String... args) {
if (args.length < 1) {
log.error("At a minimum, the stream name is required as the first argument. The Region may be specified as the second argument.");
System.exit(1);
}
String streamName = args[0];
String region = null;
if (args.length > 1) {
region = args[1];
}
new SampleSingle(streamName, region).run();
}
private final String streamName;
private final Region region;
private final KinesisAsyncClient kinesisClient;
/**
* Constructor sets streamName and region. It also creates a KinesisClient object to send data to Kinesis.
* This KinesisClient is used to send dummy data so that the consumer has something to read; it is also used
* indirectly by the KCL to handle the consumption of the data.
*/
private SampleSingle(String streamName, String region) {
this.streamName = streamName;
this.region = Region.of(ObjectUtils.firstNonNull(region, "us-east-2"));
this.kinesisClient = KinesisClientUtil.createKinesisAsyncClient(KinesisAsyncClient.builder().region(this.region));
}
private void run() {
/**
* Sends dummy data to Kinesis. Not relevant to consuming the data with the KCL
*/
ScheduledExecutorService producerExecutor = Executors.newSingleThreadScheduledExecutor();
ScheduledFuture producerFuture = producerExecutor.scheduleAtFixedRate(this::publishRecord, 10, 1, TimeUnit.SECONDS);
/**
* Sets up configuration for the KCL, including DynamoDB and CloudWatch dependencies. The final argument, a
* ShardRecordProcessorFactory, is where the logic for record processing lives, and is located in a private
* class below.
*/
DynamoDbAsyncClient dynamoClient = DynamoDbAsyncClient.builder().region(region).build();
CloudWatchAsyncClient cloudWatchClient = CloudWatchAsyncClient.builder().region(region).build();
ConfigsBuilder configsBuilder = new ConfigsBuilder(streamName, streamName, kinesisClient, dynamoClient, cloudWatchClient, UUID.randomUUID().toString(), new SampleRecordProcessorFactory());
/**
* The Scheduler (also called Worker in earlier versions of the KCL) is the entry point to the KCL. This
* instance is configured with defaults provided by the ConfigsBuilder.
*/
Scheduler scheduler = new Scheduler(
configsBuilder.checkpointConfig(),
configsBuilder.coordinatorConfig(),
configsBuilder.leaseManagementConfig(),
configsBuilder.lifecycleConfig(),
configsBuilder.metricsConfig(),
configsBuilder.processorConfig(),
configsBuilder.retrievalConfig().retrievalSpecificConfig(new PollingConfig(streamName, kinesisClient))
);
/**
* Kickoff the Scheduler. Record processing of the stream of dummy data will continue indefinitely
* until an exit is triggered.
*/
Thread schedulerThread = new Thread(scheduler);
schedulerThread.setDaemon(true);
schedulerThread.start();
/**
* Allows termination of app by pressing Enter.
*/
System.out.println("Press enter to shutdown");
BufferedReader reader = new BufferedReader(new InputStreamReader(System.in));
try {
reader.readLine();
} catch (IOException ioex) {
log.error("Caught exception while waiting for confirm. Shutting down.", ioex);
}
/**
* Stops sending dummy data.
*/
log.info("Cancelling producer and shutting down executor.");
producerFuture.cancel(true);
producerExecutor.shutdownNow();
/**
* Stops consuming data. Finishes processing the current batch of data already received from Kinesis
* before shutting down.
*/
Future gracefulShutdownFuture = scheduler.startGracefulShutdown();
log.info("Waiting up to 20 seconds for shutdown to complete.");
try {
gracefulShutdownFuture.get(20, TimeUnit.SECONDS);
} catch (InterruptedException e) {
log.info("Interrupted while waiting for graceful shutdown. Continuing.");
} catch (ExecutionException e) {
log.error("Exception while executing graceful shutdown.", e);
} catch (TimeoutException e) {
log.error("Timeout while waiting for shutdown. Scheduler may not have exited.");
}
log.info("Completed, shutting down now.");
}
/**
* Sends a single record of dummy data to Kinesis.
*/
private void publishRecord() {
PutRecordRequest request = PutRecordRequest.builder()
.partitionKey(RandomStringUtils.randomAlphabetic(5, 20))
.streamName(streamName)
.data(SdkBytes.fromByteArray(RandomUtils.nextBytes(10)))
.build();
try {
kinesisClient.putRecord(request).get();
} catch (InterruptedException e) {
log.info("Interrupted, assuming shutdown.");
} catch (ExecutionException e) {
log.error("Exception while sending data to Kinesis. Will try again next cycle.", e);
}
}
private static class SampleRecordProcessorFactory implements ShardRecordProcessorFactory {
public ShardRecordProcessor shardRecordProcessor() {
return new SampleRecordProcessor();
}
}
/**
* The implementation of the ShardRecordProcessor interface is where the heart of the record processing logic lives.
* In this example all we do to 'process' is log info about the records.
*/
private static class SampleRecordProcessor implements ShardRecordProcessor {
private static final String SHARD_ID_MDC_KEY = "ShardId";
private static final Logger log = LoggerFactory.getLogger(SampleRecordProcessor.class);
private String shardId;
/**
* Invoked by the KCL before data records are delivered to the ShardRecordProcessor instance (via
* processRecords). In this example we do nothing except some logging.
*
* @param initializationInput Provides information related to initialization.
*/
public void initialize(InitializationInput initializationInput) {
shardId = initializationInput.shardId();
MDC.put(SHARD_ID_MDC_KEY, shardId);
try {
log.info("Initializing @ Sequence: {}", initializationInput.extendedSequenceNumber());
} finally {
MDC.remove(SHARD_ID_MDC_KEY);
}
}
/**
* Handles record processing logic. The Amazon Kinesis Client Library will invoke this method to deliver
* data records to the application. In this example we simply log our records.
*
* @param processRecordsInput Provides the records to be processed as well as information and capabilities
* related to them (e.g. checkpointing).
*/
public void processRecords(ProcessRecordsInput processRecordsInput) {
MDC.put(SHARD_ID_MDC_KEY, shardId);
try {
log.info("Processing {} record(s)", processRecordsInput.records().size());
processRecordsInput.records().forEach(r -> log.info("Processing record pk: {} -- Seq: {}", r.partitionKey(), r.sequenceNumber()));
} catch (Throwable t) {
log.error("Caught throwable while processing records. Aborting.");
Runtime.getRuntime().halt(1);
} finally {
MDC.remove(SHARD_ID_MDC_KEY);
}
}
/** Called when the lease tied to this record processor has been lost. Once the lease has been lost,
* the record processor can no longer checkpoint.
*
* @param leaseLostInput Provides access to functions and data related to the loss of the lease.
*/
public void leaseLost(LeaseLostInput leaseLostInput) {
MDC.put(SHARD_ID_MDC_KEY, shardId);
try {
log.info("Lost lease, so terminating.");
} finally {
MDC.remove(SHARD_ID_MDC_KEY);
}
}
/**
* Called when all data on this shard has been processed. Checkpointing must occur in the method for record
* processing to be considered complete; an exception will be thrown otherwise.
*
* @param shardEndedInput Provides access to a checkpointer method for completing processing of the shard.
*/
public void shardEnded(ShardEndedInput shardEndedInput) {
MDC.put(SHARD_ID_MDC_KEY, shardId);
try {
log.info("Reached shard end checkpointing.");
shardEndedInput.checkpointer().checkpoint();
} catch (ShutdownException | InvalidStateException e) {
log.error("Exception while checkpointing at shard end. Giving up.", e);
} finally {
MDC.remove(SHARD_ID_MDC_KEY);
}
}
/**
* Invoked when Scheduler has been requested to shut down (i.e. we decide to stop running the app by pressing
* Enter). 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) {
MDC.put(SHARD_ID_MDC_KEY, shardId);
try {
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);
} finally {
MDC.remove(SHARD_ID_MDC_KEY);
}
}
}
}
```
# Python で Kinesis クライアントライブラリコンシューマーを開発する
**重要**
Amazon Kinesis Client Library (KCL) バージョン 1.x および 2.x は古くなっています。KCL 1.x は 2026 年 1 月 30 日にサポートが終了します。2026 年 1 月 30 日より前に、バージョン 1.x を使用して KCL アプリケーションを最新の KCL バージョンに移行することを**強くお勧めします**。最新の KCL バージョンを確認するには、[GitHub のAmazon Kinesis Client Library](https://github.com/awslabs/amazon-kinesis-client)ページを参照してください。最新の KCL バージョンの詳細については、[Kinesis Client Library を使用する](kcl.md) を参照してください。KCL 1.x から KCL 3.x への移行については、「[KCL 1.x から KCL 3.x への移行](kcl-migration-1-3.md)」を参照してください。
Kinesis Data Streams のデータを処理するアプリケーションを構築するには Kinesis Client Library (KCL) を使用できます。Kinesis Client Library は、複数の言語で使用できます。このトピックでは、Python について説明します。
KCL は Java ライブラリであり、Java 以外の言語のサポートは、*MultiLangDaemon* と呼ばれる多言語インターフェースを使用して提供されます。このデーモンは Java ベースで、Java 以外の KCL 言語を使用しているときにバックグラウンドで実行されます。そのため、KCL for Python をインストールして、コンシューマーアプリケーションをすべて Python で書く場合でも、MultiLangDaemon を使用するために、Java をシステムにインストールする必要があります。さらに、MultiLangDaemon には、接続先の AWS リージョンなど、ユースケースに合わせてカスタマイズする必要があるデフォルト設定があります。GitHub の MultiLangDaemon の詳細については、[KCL MultiLangDaemon project](https://github.com/awslabs/amazon-kinesis-client/tree/v1.x/src/main/java/com/amazonaws/services/kinesis/multilang)のページを参照してください。
GitHub から Python KCL をダウンロードするには、[Kinesis Client Library (Python)](https://github.com/awslabs/amazon-kinesis-client-python) にアクセスしてください。Python KCL コンシューマーアプリケーションのサンプルコードをダウンロードするには、GitHub で[KCL for Python sample project](https://github.com/awslabs/amazon-kinesis-client-python/tree/master/samples)ページにアクセスしてください。
Python で KCL コンシューマーアプリケーションを実装する場合は、次のタスクを完了する必要があります。
**Topics**
+ [RecordProcessor クラスのメソッドを実装する](#kinesis-record-processor-implementation-interface-py)
+ [設定プロパティを変更する](#kinesis-record-processor-initialization-py)
## RecordProcessor クラスのメソッドを実装する
`RecordProcess` クラスでは、`RecordProcessorBase` クラスを拡張して次のメソッドを実装する必要があります。
```
initialize
process_records
shutdown_requested
```
このサンプルでは、開始点として使用できる実装を提供しています。
```
#!/usr/bin/env python
# Copyright 2014-2015 Amazon.com, Inc. or its affiliates. All Rights Reserved.
#
# Licensed under the Amazon Software License (the "License").
# You may not use this file except in compliance with the License.
# A copy of the License is located at
#
# http://aws.amazon.com/asl/
#
# or in the "license" file accompanying this file. This file is distributed
# on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
# express or implied. See the License for the specific language governing
# permissions and limitations under the License.
from __future__ import print_function
import sys
import time
from amazon_kclpy import kcl
from amazon_kclpy.v3 import processor
class RecordProcessor(processor.RecordProcessorBase):
"""
A RecordProcessor processes data from a shard in a stream. Its methods will be called with this pattern:
* initialize will be called once
* process_records will be called zero or more times
* shutdown will be called if this MultiLangDaemon instance loses the lease to this shard, or the shard ends due
a scaling change.
"""
def __init__(self):
self._SLEEP_SECONDS = 5
self._CHECKPOINT_RETRIES = 5
self._CHECKPOINT_FREQ_SECONDS = 60
self._largest_seq = (None, None)
self._largest_sub_seq = None
self._last_checkpoint_time = None
def log(self, message):
sys.stderr.write(message)
def initialize(self, initialize_input):
"""
Called once by a KCLProcess before any calls to process_records
:param amazon_kclpy.messages.InitializeInput initialize_input: Information about the lease that this record
processor has been assigned.
"""
self._largest_seq = (None, None)
self._last_checkpoint_time = time.time()
def checkpoint(self, checkpointer, sequence_number=None, sub_sequence_number=None):
"""
Checkpoints with retries on retryable exceptions.
:param amazon_kclpy.kcl.Checkpointer checkpointer: the checkpointer provided to either process_records
or shutdown
:param str or None sequence_number: the sequence number to checkpoint at.
:param int or None sub_sequence_number: the sub sequence number to checkpoint at.
"""
for n in range(0, self._CHECKPOINT_RETRIES):
try:
checkpointer.checkpoint(sequence_number, sub_sequence_number)
return
except kcl.CheckpointError as e:
if 'ShutdownException' == e.value:
#
# A ShutdownException indicates that this record processor should be shutdown. This is due to
# some failover event, e.g. another MultiLangDaemon has taken the lease for this shard.
#
print('Encountered shutdown exception, skipping checkpoint')
return
elif 'ThrottlingException' == e.value:
#
# A ThrottlingException indicates that one of our dependencies is is over burdened, e.g. too many
# dynamo writes. We will sleep temporarily to let it recover.
#
if self._CHECKPOINT_RETRIES - 1 == n:
sys.stderr.write('Failed to checkpoint after {n} attempts, giving up.\n'.format(n=n))
return
else:
print('Was throttled while checkpointing, will attempt again in {s} seconds'
.format(s=self._SLEEP_SECONDS))
elif 'InvalidStateException' == e.value:
sys.stderr.write('MultiLangDaemon reported an invalid state while checkpointing.\n')
else: # Some other error
sys.stderr.write('Encountered an error while checkpointing, error was {e}.\n'.format(e=e))
time.sleep(self._SLEEP_SECONDS)
def process_record(self, data, partition_key, sequence_number, sub_sequence_number):
"""
Called for each record that is passed to process_records.
:param str data: The blob of data that was contained in the record.
:param str partition_key: The key associated with this recod.
:param int sequence_number: The sequence number associated with this record.
:param int sub_sequence_number: the sub sequence number associated with this record.
"""
####################################
# Insert your processing logic here
####################################
self.log("Record (Partition Key: {pk}, Sequence Number: {seq}, Subsequence Number: {sseq}, Data Size: {ds}"
.format(pk=partition_key, seq=sequence_number, sseq=sub_sequence_number, ds=len(data)))
def should_update_sequence(self, sequence_number, sub_sequence_number):
"""
Determines whether a new larger sequence number is available
:param int sequence_number: the sequence number from the current record
:param int sub_sequence_number: the sub sequence number from the current record
:return boolean: true if the largest sequence should be updated, false otherwise
"""
return self._largest_seq == (None, None) or sequence_number > self._largest_seq[0] or \
(sequence_number == self._largest_seq[0] and sub_sequence_number > self._largest_seq[1])
def process_records(self, process_records_input):
"""
Called by a KCLProcess with a list of records to be processed and a checkpointer which accepts sequence numbers
from the records to indicate where in the stream to checkpoint.
:param amazon_kclpy.messages.ProcessRecordsInput process_records_input: the records, and metadata about the
records.
"""
try:
for record in process_records_input.records:
data = record.binary_data
seq = int(record.sequence_number)
sub_seq = record.sub_sequence_number
key = record.partition_key
self.process_record(data, key, seq, sub_seq)
if self.should_update_sequence(seq, sub_seq):
self._largest_seq = (seq, sub_seq)
#
# Checkpoints every self._CHECKPOINT_FREQ_SECONDS seconds
#
if time.time() - self._last_checkpoint_time > self._CHECKPOINT_FREQ_SECONDS:
self.checkpoint(process_records_input.checkpointer, str(self._largest_seq[0]), self._largest_seq[1])
self._last_checkpoint_time = time.time()
except Exception as e:
self.log("Encountered an exception while processing records. Exception was {e}\n".format(e=e))
def lease_lost(self, lease_lost_input):
self.log("Lease has been lost")
def shard_ended(self, shard_ended_input):
self.log("Shard has ended checkpointing")
shard_ended_input.checkpointer.checkpoint()
def shutdown_requested(self, shutdown_requested_input):
self.log("Shutdown has been requested, checkpointing.")
shutdown_requested_input.checkpointer.checkpoint()
if __name__ == "__main__":
kcl_process = kcl.KCLProcess(RecordProcessor())
kcl_process.run()
```
## 設定プロパティを変更する
このサンプルでは、次のスクリプトに示すように、設定プロパティのデフォルト値を提供します。これらのプロパティを独自の値にオーバーライドできます。
```
# The script that abides by the multi-language protocol. This script will
# be executed by the MultiLangDaemon, which will communicate with this script
# over STDIN and STDOUT according to the multi-language protocol.
executableName = sample_kclpy_app.py
# The name of an Amazon Kinesis stream to process.
streamName = words
# Used by the KCL as the name of this application. Will be used as the name
# of an Amazon DynamoDB table which will store the lease and checkpoint
# information for workers with this application name
applicationName = PythonKCLSample
# Users can change the credentials provider the KCL will use to retrieve credentials.
# The DefaultAWSCredentialsProviderChain checks several other providers, which is
# described here:
# http://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/com/amazonaws/auth/DefaultAWSCredentialsProviderChain.html
AWSCredentialsProvider = DefaultAWSCredentialsProviderChain
# Appended to the user agent of the KCL. Does not impact the functionality of the
# KCL in any other way.
processingLanguage = python/2.7
# Valid options at TRIM_HORIZON or LATEST.
# See http://docs.aws.amazon.com/kinesis/latest/APIReference/API_GetShardIterator.html#API_GetShardIterator_RequestSyntax
initialPositionInStream = TRIM_HORIZON
# The following properties are also available for configuring the KCL Worker that is created
# by the MultiLangDaemon.
# The KCL defaults to us-east-1
#regionName = us-east-1
# Fail over time in milliseconds. A worker which does not renew it's lease within this time interval
# will be regarded as having problems and it's shards will be assigned to other workers.
# For applications that have a large number of shards, this msy be set to a higher number to reduce
# the number of DynamoDB IOPS required for tracking leases
#failoverTimeMillis = 10000
# A worker id that uniquely identifies this worker among all workers using the same applicationName
# If this isn't provided a MultiLangDaemon instance will assign a unique workerId to itself.
#workerId =
# Shard sync interval in milliseconds - e.g. wait for this long between shard sync tasks.
#shardSyncIntervalMillis = 60000
# Max records to fetch from Kinesis in a single GetRecords call.
#maxRecords = 10000
# Idle time between record reads in milliseconds.
#idleTimeBetweenReadsInMillis = 1000
# Enables applications flush/checkpoint (if they have some data "in progress", but don't get new data for while)
#callProcessRecordsEvenForEmptyRecordList = false
# Interval in milliseconds between polling to check for parent shard completion.
# Polling frequently will take up more DynamoDB IOPS (when there are leases for shards waiting on
# completion of parent shards).
#parentShardPollIntervalMillis = 10000
# Cleanup leases upon shards completion (don't wait until they expire in Kinesis).
# Keeping leases takes some tracking/resources (e.g. they need to be renewed, assigned), so by default we try
# to delete the ones we don't need any longer.
#cleanupLeasesUponShardCompletion = true
# Backoff time in milliseconds for Amazon Kinesis Client Library tasks (in the event of failures).
#taskBackoffTimeMillis = 500
# Buffer metrics for at most this long before publishing to CloudWatch.
#metricsBufferTimeMillis = 10000
# Buffer at most this many metrics before publishing to CloudWatch.
#metricsMaxQueueSize = 10000
# KCL will validate client provided sequence numbers with a call to Amazon Kinesis before checkpointing for calls
# to RecordProcessorCheckpointer#checkpoint(String) by default.
#validateSequenceNumberBeforeCheckpointing = true
# The maximum number of active threads for the MultiLangDaemon to permit.
# If a value is provided then a FixedThreadPool is used with the maximum
# active threads set to the provided value. If a non-positive integer or no
# value is provided a CachedThreadPool is used.
#maxActiveThreads = 0
```
### アプリケーション名
KCL には、複数のアプリケーション間、および同じリージョン内の Amazon DynamoDB テーブル間で一意のアプリケーションが必要です。次のようにアプリケーション名の設定値を使用します。
+ このアプリケーション名と関連付けられたワーカーはすべて、同じストリーム上で連携して処理しているとみなされます。これらのワーカーは複数のインスタンス間に分散している場合があります。同じアプリケーションコードの追加のインスタンスを実行するときに、アプリケーション名が異なる場合、KCL は 2 番目のインスタンスを、同じストリームで動作するまったく別のアプリケーションと見なします。
+ KCL はアプリケーション名を使用して DynamoDB テーブルを作成し、このテーブルを使用してアプリケーションの状態情報 (チェックポイントやワーカーとシャードのマッピングなど) を保存します。各アプリケーションには、それぞれ DynamoDB テーブルがあります。詳細については、「[リーステーブルを使用して KCL コンシューマーアプリケーションによって処理されたシャードを追跡する](shared-throughput-kcl-consumers.md#shared-throughput-kcl-consumers-leasetable)」を参照してください。
### 認証情報
デフォルトの AWS 認証情報プロバイダー[チェーンの認証情報プロバイダー](https://docs.aws.amazon.com/sdk-for-java/latest/reference/com/amazonaws/auth/DefaultAWSCredentialsProviderChain.html)の 1 つが認証情報を利用できるようにする必要があります。`AWSCredentialsProvider` プロパティを使用して認証情報プロバイダーを設定できます。Amazon EC2 インスタンスでコンシューマーアプリケーションを実行する場合は、この IAM ロールに関連付けられたアクセス許可を反映する IAM role. AWS credentials を使用してインスタンスを設定することをお勧めします。この IAM ロールは、インスタンスメタデータを介してインスタンス上のアプリケーションで使用できます。これは、EC2 インスタンスで実行されるコンシューマーアプリケーションの認証情報を管理するための最も安全な方法です。
# KCL 2.x を使用して拡張ファンアウトコンシューマーを開発する
**重要**
Amazon Kinesis Client Library (KCL) バージョン 1.x および 2.x は古くなっています。KCL 1.x は 2026 年 1 月 30 日にサポートが終了します。2026 年 1 月 30 日より前に、バージョン 1.x を使用して KCL アプリケーションを最新の KCL バージョンに移行することを**強くお勧めします**。最新の KCL バージョンを確認するには、[GitHub のAmazon Kinesis Client Library](https://github.com/awslabs/amazon-kinesis-client)ページを参照してください。最新の KCL バージョンの詳細については、[Kinesis Client Library を使用する](kcl.md) を参照してください。KCL 1.x から KCL 3.x への移行については、「[KCL 1.x から KCL 3.x への移行](kcl-migration-1-3.md)」を参照してください。
Amazon Kinesis Data Streams で*拡張ファンアウト*を使用するコンシューマーは、シャードあたり 1 秒間に最大 2 MB のデータの専用スループットで、データストリームからレコードを受け取ることができます。このタイプのコンシューマーは、ストリームからデータを受け取っている他のコンシューマーと競合する必要はありません。詳細については、[専用スループットを備えた拡張ファンアウトを開発する](enhanced-consumers.md)を参照してください。
拡張ファンアウトを使用してストリームからデータを受け取るアプリケーションを開発するには、バージョン 2.0 以降の Kinesis Client Library (KCL) を使用できます。KCL は、アプリケーションをストリームのすべてのシャードに自動的にサブスクライブし、コンシューマーアプリケーションがシャードあたり 2 MB/秒のスループット値で読み取ることができるようにします。拡張ファンアウトをオンにせずに KCL を使用する場合は、[Kinesis Client Library 2.0 を使用したコンシューマーの開発](https://docs.aws.amazon.com/streams/latest/dev/developing-consumers-with-kcl-v2.html)を参照してください。
**Topics**
+ [Java で KCL 2.x を使用して拡張ファンアウトコンシューマーを開発する](building-enhanced-consumers-kcl-java.md)
# Java で KCL 2.x を使用して拡張ファンアウトコンシューマーを開発する
**重要**
Amazon Kinesis Client Library (KCL) バージョン 1.x および 2.x は古くなっています。KCL 1.x は 2026 年 1 月 30 日にサポートが終了します。2026 年 1 月 30 日より前に、バージョン 1.x を使用して KCL アプリケーションを最新の KCL バージョンに移行することを**強くお勧めします**。最新の KCL バージョンを確認するには、[GitHub のAmazon Kinesis Client Library](https://github.com/awslabs/amazon-kinesis-client)ページを参照してください。最新の KCL バージョンの詳細については、[Kinesis Client Library を使用する](kcl.md) を参照してください。KCL 1.x から KCL 3.x への移行については、「[KCL 1.x から KCL 3.x への移行](kcl-migration-1-3.md)」を参照してください。
拡張ファンアウトを使用してストリームからデータを受け取るアプリケーションを Amazon Kinesis Data Streams で開発するには、バージョン 2.0 以降の Kinesis Client Library (KCL) を使用できます。次のコードは、`ProcessorFactory` および `RecordProcessor` の Java のサンプル実装を示しています。
`KinesisClientUtil` を使用して `KinesisAsyncClient` を作成し、`KinesisAsyncClient` で `maxConcurrency` を設定することをお勧めします。
**重要**
すべてのリースと `KinesisAsyncClient` の追加使用のための十分な高い `maxConcurrency` を持つよう `KinesisAsyncClient` を設定しないと、Amazon Kinesis Client で非常に大きなレイテンシーが発生する可能性があります。
```
/*
* Copyright 2019 Amazon.com, Inc. or its affiliates. All Rights Reserved.
*
* Licensed under the Amazon Software License (the "License").
* You may not use this file except in compliance with the License.
* A copy of the License is located at
*
* http://aws.amazon.com/asl/
*
* or in the "license" file accompanying this file. This file is distributed
* on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
* express or implied. See the License for the specific language governing
* permissions and limitations under the License.
*/
/*
* Copyright 2019 Amazon.com, Inc. or its affiliates. All Rights Reserved.
*
* Licensed under the Apache License, Version 2.0 (the "License").
* You may not use this file except in compliance with the License.
* A copy of the License is located at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* or in the "license" file accompanying this file. This file is distributed
* on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
* express or implied. See the License for the specific language governing
* permissions and limitations under the License.
*/
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.UUID;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.apache.commons.lang3.ObjectUtils;
import org.apache.commons.lang3.RandomStringUtils;
import org.apache.commons.lang3.RandomUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;
import software.amazon.awssdk.core.SdkBytes;
import software.amazon.awssdk.regions.Region;
import software.amazon.awssdk.services.cloudwatch.CloudWatchAsyncClient;
import software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClient;
import software.amazon.awssdk.services.kinesis.KinesisAsyncClient;
import software.amazon.awssdk.services.kinesis.model.PutRecordRequest;
import software.amazon.kinesis.common.ConfigsBuilder;
import software.amazon.kinesis.common.KinesisClientUtil;
import software.amazon.kinesis.coordinator.Scheduler;
import software.amazon.kinesis.exceptions.InvalidStateException;
import software.amazon.kinesis.exceptions.ShutdownException;
import software.amazon.kinesis.lifecycle.events.InitializationInput;
import software.amazon.kinesis.lifecycle.events.LeaseLostInput;
import software.amazon.kinesis.lifecycle.events.ProcessRecordsInput;
import software.amazon.kinesis.lifecycle.events.ShardEndedInput;
import software.amazon.kinesis.lifecycle.events.ShutdownRequestedInput;
import software.amazon.kinesis.processor.ShardRecordProcessor;
import software.amazon.kinesis.processor.ShardRecordProcessorFactory;
public class SampleSingle {
private static final Logger log = LoggerFactory.getLogger(SampleSingle.class);
public static void main(String... args) {
if (args.length < 1) {
log.error("At a minimum, the stream name is required as the first argument. The Region may be specified as the second argument.");
System.exit(1);
}
String streamName = args[0];
String region = null;
if (args.length > 1) {
region = args[1];
}
new SampleSingle(streamName, region).run();
}
private final String streamName;
private final Region region;
private final KinesisAsyncClient kinesisClient;
private SampleSingle(String streamName, String region) {
this.streamName = streamName;
this.region = Region.of(ObjectUtils.firstNonNull(region, "us-east-2"));
this.kinesisClient = KinesisClientUtil.createKinesisAsyncClient(KinesisAsyncClient.builder().region(this.region));
}
private void run() {
ScheduledExecutorService producerExecutor = Executors.newSingleThreadScheduledExecutor();
ScheduledFuture producerFuture = producerExecutor.scheduleAtFixedRate(this::publishRecord, 10, 1, TimeUnit.SECONDS);
DynamoDbAsyncClient dynamoClient = DynamoDbAsyncClient.builder().region(region).build();
CloudWatchAsyncClient cloudWatchClient = CloudWatchAsyncClient.builder().region(region).build();
ConfigsBuilder configsBuilder = new ConfigsBuilder(streamName, streamName, kinesisClient, dynamoClient, cloudWatchClient, UUID.randomUUID().toString(), new SampleRecordProcessorFactory());
Scheduler scheduler = new Scheduler(
configsBuilder.checkpointConfig(),
configsBuilder.coordinatorConfig(),
configsBuilder.leaseManagementConfig(),
configsBuilder.lifecycleConfig(),
configsBuilder.metricsConfig(),
configsBuilder.processorConfig(),
configsBuilder.retrievalConfig()
);
Thread schedulerThread = new Thread(scheduler);
schedulerThread.setDaemon(true);
schedulerThread.start();
System.out.println("Press enter to shutdown");
BufferedReader reader = new BufferedReader(new InputStreamReader(System.in));
try {
reader.readLine();
} catch (IOException ioex) {
log.error("Caught exception while waiting for confirm. Shutting down.", ioex);
}
log.info("Cancelling producer, and shutting down executor.");
producerFuture.cancel(true);
producerExecutor.shutdownNow();
Future gracefulShutdownFuture = scheduler.startGracefulShutdown();
log.info("Waiting up to 20 seconds for shutdown to complete.");
try {
gracefulShutdownFuture.get(20, TimeUnit.SECONDS);
} catch (InterruptedException e) {
log.info("Interrupted while waiting for graceful shutdown. Continuing.");
} catch (ExecutionException e) {
log.error("Exception while executing graceful shutdown.", e);
} catch (TimeoutException e) {
log.error("Timeout while waiting for shutdown. Scheduler may not have exited.");
}
log.info("Completed, shutting down now.");
}
private void publishRecord() {
PutRecordRequest request = PutRecordRequest.builder()
.partitionKey(RandomStringUtils.randomAlphabetic(5, 20))
.streamName(streamName)
.data(SdkBytes.fromByteArray(RandomUtils.nextBytes(10)))
.build();
try {
kinesisClient.putRecord(request).get();
} catch (InterruptedException e) {
log.info("Interrupted, assuming shutdown.");
} catch (ExecutionException e) {
log.error("Exception while sending data to Kinesis. Will try again next cycle.", e);
}
}
private static class SampleRecordProcessorFactory implements ShardRecordProcessorFactory {
public ShardRecordProcessor shardRecordProcessor() {
return new SampleRecordProcessor();
}
}
private static class SampleRecordProcessor implements ShardRecordProcessor {
private static final String SHARD_ID_MDC_KEY = "ShardId";
private static final Logger log = LoggerFactory.getLogger(SampleRecordProcessor.class);
private String shardId;
public void initialize(InitializationInput initializationInput) {
shardId = initializationInput.shardId();
MDC.put(SHARD_ID_MDC_KEY, shardId);
try {
log.info("Initializing @ Sequence: {}", initializationInput.extendedSequenceNumber());
} finally {
MDC.remove(SHARD_ID_MDC_KEY);
}
}
public void processRecords(ProcessRecordsInput processRecordsInput) {
MDC.put(SHARD_ID_MDC_KEY, shardId);
try {
log.info("Processing {} record(s)", processRecordsInput.records().size());
processRecordsInput.records().forEach(r -> log.info("Processing record pk: {} -- Seq: {}", r.partitionKey(), r.sequenceNumber()));
} catch (Throwable t) {
log.error("Caught throwable while processing records. Aborting.");
Runtime.getRuntime().halt(1);
} finally {
MDC.remove(SHARD_ID_MDC_KEY);
}
}
public void leaseLost(LeaseLostInput leaseLostInput) {
MDC.put(SHARD_ID_MDC_KEY, shardId);
try {
log.info("Lost lease, so terminating.");
} finally {
MDC.remove(SHARD_ID_MDC_KEY);
}
}
public void shardEnded(ShardEndedInput shardEndedInput) {
MDC.put(SHARD_ID_MDC_KEY, shardId);
try {
log.info("Reached shard end checkpointing.");
shardEndedInput.checkpointer().checkpoint();
} catch (ShutdownException | InvalidStateException e) {
log.error("Exception while checkpointing at shard end. Giving up.", e);
} finally {
MDC.remove(SHARD_ID_MDC_KEY);
}
}
public void shutdownRequested(ShutdownRequestedInput shutdownRequestedInput) {
MDC.put(SHARD_ID_MDC_KEY, shardId);
try {
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);
} finally {
MDC.remove(SHARD_ID_MDC_KEY);
}
}
}
}
```