イベント処理のためのパーティション負荷分散

パーティション負荷分散は、アプリケーションの複数のインスタンスにイベント処理ワークロードを分散するAzure Event Hubsの手法です。 イベント プロセッサ クライアントは、パーティションの所有権を自動的に管理し、アクティブなすべてのコンシューマー インスタンス間の作業の分散を調整します。

新しい SDK バージョン (5.0 以降) では、EventProcessorClient (.NETとJava) または EventHubConsumerClient (Python および JavaScript) によって負荷分散が自動的に処理されます。 関心のあるイベントをサブスクライブするには、イベント ハンドラーを登録します。

この記事では、クライアント アプリケーションの複数のインスタンスを使用して、イベント ハブからイベントを読み取るためのサンプル シナリオについて説明します。 また、パーティションの所有権、チェックポイント処理、負荷分散などの主要な概念についても説明します。

サンプル シナリオ

シナリオの例として、10 万件の家を監視するホーム セキュリティ企業を考えてみましょう。 この会社では、各家庭に設置された動体検知器、ドアや窓の開閉センサー、ガラス破損検知器などのさまざまなセンサーから常にデータを取得しています。 この会社では、住民がほぼリアルタイムで自宅の様子を監視できる Web サイトを開設しています。

各センサーにより、データがイベント ハブにプッシュされます。 イベント ハブは、16 個のパーティションで構成されます。 使用側では、これらのイベントを読み取り、統合し (フィルター、集計など)、集計をストレージ BLOB にダンプし、ユーザー フレンドリな Web ページに投影できるメカニズムが必要です。

コンシューマー アプリケーション

分散環境でコンシューマーを設計する場合、シナリオで次の要件を扱う必要があります。

• スケール: 複数のコンシューマーを作成します。それぞれのコンシューマーは、いくつかの Event Hubs のパーティションからの読み取りの所有権を保持します。
• 負荷分散: コンシューマーを動的に増減します。 たとえば、新しいセンサーの種類 (たとえば、一酸化炭素検知器) が各家庭に追加されると、イベントの数が増加します。 その場合は、オペレーター (人間) がコンシューマー インスタンスの数を増やします。 すると、コンシューマーのプールが所有するパーティションの数をリバランスし、新しく追加されたコンシューマーと負荷を共有することができます。
• 失敗時のシームレスな再開: ホストとなっている仮想マシンが突然クラッシュしたなどの理由でコンシューマー (コンシューマー A) が失敗した場合、コンシューマー A が所有しているパーティションを他のコンシューマーが選択して続行できます。 また、"チェックポイント" または "オフセット" と呼ばれる継続ポイントは、コンシューマー A が失敗した正確なポイントであるか、その少し前のポイントである必要があります。
• イベントの使用: 前の 3 つのポイントではコンシューマーの管理を扱っていますが、イベントを使用して実用的な操作を行うには、コードが必要です。 たとえば、データを集計して BLOB ストレージにアップロードします。

イベント プロセッサまたはコンシューマー クライアント

これらの要件を満たすために独自のソリューションを構築する必要はありません。 この機能は、Azure Event Hubs SDK によって提供されます。 .NET SDK またはJava SDK では、イベント プロセッサ クライアント (EventProcessorClient) を使用します。 Python SDK と JavaScript SDK では、EventHubConsumerClient を使用します。 古いバージョンの SDK では、イベント プロセッサ ホスト (EventProcessorHost) でこれらの機能がサポートされました。

ほとんどの実稼働シナリオでは、イベントの読み取りと処理にイベント プロセッサ クライアントを使用します。 プロセッサ クライアントは、高性能かつフォールトトレラントな方法でイベント ハブのすべてのパーティションにおけるイベント処理能力を提供しながら、進行状況を管理するためのチェックポイント機能も備えています。 イベント プロセッサ クライアントは、特定のイベント ハブ用にコンシューマー グループのコンテキスト内で協調的に動作できます。 インスタンスがグループで利用可能または利用不可になると、クライアントは作業の分散と負荷分散を自動的に管理します。

パーティションの所有権

通常、イベント プロセッサ インスタンスは、1 つまたは複数のパーティションからのイベントを所有および処理します。 システムは、イベント ハブとコンシューマー グループの組み合わせに関連付けられているすべてのアクティブなイベント プロセッサ インスタンス間でパーティションの所有権を均等に分散します。

各イベント プロセッサは、チェックポイント ストア内のエントリを追加または更新することによって、一意の識別子を持ち、パーティションの所有権を要求します。 すべてのイベント プロセッサ インスタンスは、このストアと定期的に通信して、独自の処理状態を更新し、他のアクティブなインスタンスについて学習します。 システムはこのデータを使用して、アクティブなプロセッサ間の負荷を分散します。 新しいインスタンスは、処理プールに参加してスケール アップできます。 障害が原因で、またはスケール ダウンのためにインスタンスがダウンすると、システムはパーティションの所有権を他のアクティブなプロセッサに適切に転送します。

チェックポイント ストアのパーティションの所有権レコードでは、Event Hubs 名前空間、イベント ハブ名、コンシューマー グループ、イベント プロセッサ識別子 (所有者とも呼ばれます)、パーティション ID、および最終変更時刻が追跡されます。

各イベント プロセッサ インスタンスは、パーティションの所有権を取得し、最後に認識されたチェックポイントからパーティションの処理を開始します。 プロセッサが失敗した場合 (VM がシャットダウン)、他のインスタンスは最後に変更された時刻を調べることでエラーを検出します。 非アクティブなインスタンスによって以前所有されていたパーティションの所有権を取得しようとする試みは、他のインスタンスによって行われます。 チェックポイント ストアを使用すると、1 つのインスタンスのみがパーティションの所有権の要求に成功することが保証されます。 そのため、特定の時点で、パーティションからイベントを受信するプロセッサが最大 1 つ存在します。

メッセージを受信する

イベント プロセッサを作成するときは、イベントとエラーを処理する関数を指定します。 イベントを処理する関数を呼び出すたびに、特定のパーティションから 1 つのイベントが配信されます。 このイベントを処理する必要があります。 コンシューマーがすべてのメッセージを少なくとも 1 回処理することを確認する場合は、再試行ロジックを使用して独自のコードを記述します。 ただし、有害メッセージについて注意してください。

比較的高速にイベントを処理します。 つまり、できる限り最小限の処理に留めます。 ストレージへの書き込みとルーティングを行う必要がある場合、2 つのコンシューマー グループを使用して 2 つのイベント プロセッサを所有することをお勧めします。

チェックポイント

チェックポイント処理とは、イベント プロセッサがパーティション内の最後に正常に処理されたイベントの位置をマークまたはコミットするために使用する処理です。 通常、チェックポイントのマークは、イベントを処理する関数内で発生し、コンシューマー グループ内のパーティションごとに発生します。

イベント プロセッサがパーティションから切断された場合、別のインスタンスは、そのコンシューマー グループ内のそのパーティションの最後のプロセッサが以前にコミットしたチェックポイントでパーティションの処理を再開できます。 プロセッサは接続の際に、このオフセットをイベント ハブに渡して、読み取りを開始する場所を指定します。 このように、チェックポイント処理を使用することで、ダウンストリーム アプリケーションごとにイベントに "完了" のマークを付けると共に、イベント プロセッサがダウンしたときに回復性をもたらすことができます。 古いデータに戻すには、このチェックポイント処理プロセスから低いオフセットを指定します。

チェックポイントは、イベントを処理済みとしてマークすると、イベントのオフセットとシーケンス番号を使用してチェックポイント ストア内のエントリを追加または更新します。 チェックポイントを更新する頻度を決定します。 正常に処理された各イベントの後に更新すると、基になっているチェックポイント ストアへの書き込み操作がトリガーされるため、パフォーマンスとコストへの影響が生じる可能性があります。 すべての単一イベントをチェックポイント処理することは、キューに格納されたメッセージング パターンを暗示しています。その場合は、イベント ハブよりも Service Bus キューの方がより適切なオプションになる可能性があります。 Event Hubs の背後にあるのは、"1 回以上" 大規模な配信を受ける、という考え方です。 ダウンストリームのシステムにべき等性を持たせることで、同じイベントが複数回受信される結果になるエラーまたは再起動から容易に復旧できます。

チェックポイント ストアとして Azure Blob Storage を使用する場合は、次の推奨事項に従います。

• コンシューマー グループごとに個別のコンテナーを使用します。 同じストレージ アカウントを使用できますが、各グループごとに 1 つのコンテナーを使用します。
• ストレージ アカウントを他の目的で使用しないでください。
• コンテナーを他の目的で使用しないでください。
• デプロイされたアプリケーションと同じリージョンにストレージ アカウントを作成します。 アプリケーションがオンプレミスの場合は、可能な中で最も近いリージョンを選択することを試みてください。

Azure portal の [ストレージ アカウント] ページの [Blob service] セクションで、次の設定が無効になっていることを確認してください。

• 階層型名前空間
• BLOB の論理的な削除
• バージョン管理

スレッドの安全性とプロセッサのインスタンス

既定では、イベントを処理する関数は、特定のパーティションに対して順番に呼び出されます。 後続のイベントと同じパーティションからのこの関数に対する呼び出しは、メッセージ ポンプが他のスレッドのバックグラウンドで引き続き実行されるため、バックグラウンドでキューに配置されます。 異なるパーティションからのイベントを同時に処理できます。 パーティション間でアクセスされるすべての共有状態を同期する必要があります。

関連するコンテンツ

次のクイック スタートを参照してください。

• .NET Core
• ジャワ
• パイソン
• JavaScript