Amazon Kinesis Data Streams
開発者ガイド

.NET での Kinesis Client Library コンシューマーの開発

Kinesis データストリームのデータを処理するアプリケーションを構築するには、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」のページを参照してください。

GitHub から .NET KCL をダウンロードするには、Kinesis Client Library (.NET) にアクセスしてください。.NET KCL コンシューマーアプリケーションのサンプルコードをダウンロードするには、GitHub で「KCL for .NET sample consumer project」のページにアクセスしてください。

.NET で KCLコンシューマーアプリケーションを実装する場合は、次のタスクを完了する必要があります。

IRecordProcessor クラスのメソッドを実装する

コンシューマーでは、IRecordProcessor の次のメソッドを実装する必要があります。出発点として使用できる実装がサンプルコンシューマーに提供されています (SampleRecordProcessorSampleConsumer/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 回以上処理されるためです。特定のシャードが複数のワーカーによって処理される可能性がある場合の詳細については、「リシャーディング、拡張、並列処理」を参照してください。

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 テーブルがあります。詳細については、「Amazon Kinesis Data Streams Application の状態の追跡」を参照してください。

認証情報の設定

デフォルトの認証情報プロバイダチェーンのいずれかの認証情報プロバイダで AWS の認証情報を使用できるようにする必要があります。AWSCredentialsProvider プロパティを使用して認証情報プロバイダーを設定できます。sample.properties では、デフォルトの認証情報プロバイダーチェーンのいずれかの認証情報プロバイダーに対して、ユーザーの認証情報を使用可能にする必要があります。EC2 インスタンスでコンシューマーアプリケーションを実行している場合は、インスタンスに IAM ロールを設定することをお勧めします。この IAM ロールに関連付けられたアクセス許可を反映する AWS の認証情報は、インスタンスメタデータを通じて、インスタンス上のアプリケーションで使用できるようになります。これは、EC2 インスタンスで実行されるコンシューマーの認証情報を管理するための最も安全な方法です。

サンプルのプロパティファイルでは、AmazonKinesisSampleConsumer.cs で指定されているレコードプロセッサを使用して「words」という Kinesis data stream を処理するように KCL を設定します。