[翻訳] Data Prepper のエンドツーエンド確認応答

本記事は OpenSearch Project Blog に投稿された以下の記事を日本語に翻訳したものです。

https://opensearch.org/blog/end-to-end-acknowledgements-in-data-prepper/

Data Prepper は、OpenSearch クラスターにデータを取り込むために使用されるオープンソースのデータコレクターです。Data Prepper のインメモリバッファは高速なスループットを実現しますが、ユーザーはデータの耐久性向上、特に Data Prepper が受信したデータが目的の シンク に到達したことの確認も求めています。エンドツーエンド確認応答は、Data Prepper にこの機能を提供します。

必要性: データ耐久性の向上

Data Prepper のメンテナーとそのチームは、データ耐久性に対する一般的な課題が取り込みパイプラインの信頼性であることを観察してきました。例えば、クラスターへの一時的な負荷やスケーリング不足により OpenSearch クラスターがデータを受信できない場合、Data Prepper は宛先シンクにデータを送信できません。さらに、Data Prepper がメモリやその他のシステムリソースを使い果たしたり、Data Prepper を実行しているハードウェアが故障したりすると、取り込み中にデータが失われる可能性があります。

これらの課題を解決するために、メンテナーとそのチームはデータソース自体を考慮する必要がありました。例えば、Amazon S3 ソースは高い耐久性を持つストアからデータを読み取ることができます。この機能を観察した結果、Data Prepper が OpenSearch へのデータ配信に失敗した場合、Data Prepper は Amazon S3 からデータを再読み取りできることがわかりました。ただし、これには消費する S3 オブジェクトが利用可能であることを Data Prepper に通知する SQS メッセージを削除する前に、データが書き込まれたことを知る必要があります。

ソリューション: エンドツーエンド確認応答

Data Prepper は、エンドツーエンド確認応答を使用してデータ耐久性を提供します。Data Prepper ソースがエンドツーエンド確認応答を使用するように設定されている場合、ソースはデータがシンクに正常に配信されたときにのみ通知を受けます。ソースがエンドツーエンド確認応答を受信すると、シンクに正常に配信された SQS メッセージの削除や、Kafka ソースの場合のコミットオフセットの増分など、適切なアクションを実行できます。エンドツーエンド確認応答が受信されない場合、ソースは操作を再試行するか、外部ソースに失敗を通知することがあります。以下の図に、Data Prepper ソースがエンドツーエンド確認応答を使用するように設定されている場合の制御フローを示します。

1. Data Prepper ソースは外部ソース (Amazon S3 など) からレコードのバッチを受信します。
2. Data Prepper ソースがエンドツーエンド確認応答で設定されている場合、ソースは外部ソースから受信したレコードの各バッチに対して AcknowledgmentSet とコールバック関数を作成します。その後、ソースはレコードを Data Prepper イベントに変換します。
3. バッチ内の各イベントは AcknowledgmentSet に追加され、AcknowledgmentSet への参照がハンドルとしてイベント内に保持されます。
4. イベントは取り込みパイプラインに渡され、パイプラインは複数のプロセッサで構成され、イベント内のデータを変換、フィルタリング、エンリッチします。
5. イベントがプロセッサでドロップされた場合、イベントハンドルが解放されると AcknowledgmentSet にイベントの完了が通知されます。
6. ドロップされなかったすべてのイベントは Data Prepper シンクに送信されます。
7. Data Prepper シンクはイベントを外部シンク (OpenSearch クラスターなど) に送信します。
8. イベントが外部シンクに正常に送信されると、イベントハンドルが解放されたときに AcknowledgmentSet にイベントの完了が通知されます。
9. AcknowledgmentSet に保留中のイベントがなくなると、コールバック関数が呼び出されます。
10. コールバック関数は、レコードのバッチが正常に配信されたことを外部ソースに通知します。

すべてのイベントが外部シンクに正常に配信されない場合、AcknowledgmentSet は解放されていない一部のイベントを保持します。各 AcknowledgmentSet には有効期限があります。有効期限前にすべてのイベントが解放されない場合、AcknowledgmentSet は解放され、外部ソースにはレコードの配信が通知されません。この状況により、Data Prepper ソースまたは外部ソースがレコードの取り込みを再試行するなどの是正措置を取る可能性があります。

Data Prepper は複数のパイプラインをサポートしており、イベントは設定とイベント内のデータに基づいて異なるパイプラインにルーティングできます。さらに、各イベントは最終的に異なるシンクにデータを送信する複数のパイプラインにルーティングされる場合があります。場合によっては、イベントが複数のパイプラインまたはシンクにルーティングされるときに、同じイベントの複数のコピーが作成されないことがあります。これは、同じイベントに対して複数の確認応答が必要になるため、確認応答の追跡に課題をもたらします。この問題に対処するために、複数のパイプラインにルーティングされるイベントには参照カウントが維持されます。イベントハンドルが解放されると (ステップ 5 またはステップ 8 のいずれか)、参照カウントがデクリメントされます。イベントは完了したとみなされ、参照カウントがゼロに達したときにのみハンドルが AcknowledgmentSet から削除され、すべてのシンクへのイベントの正常な配信を示します。このアプローチにより、複数のパイプラインにルーティングされたイベントの確認応答を正確に追跡できます。

Data Prepper は、明示的な失敗を示すための否定確認応答の送信もサポートしています。コールバック関数は、確認応答のステータスが肯定か否定かを調べ、その情報に基づいて適切なアクションを実行できます。

特定のシナリオでは、Data Prepper シンクが Dead Letter Queue (DLQ) で設定されている場合、外部シンク (例: OpenSearch) に配信できないイベントは DLQ に書き込まれます。エンドツーエンド確認応答が有効な場合、(外部シンクへの配信に失敗した後) DLQ へのイベントの書き込みが成功すると、イベント配信の正常な完了とみなされます。この場合、肯定確認応答が AcknowledgmentSet に配信され、イベントの処理と処理が成功したことを示します。

今後の予定: まとめと次のステップ

エンドツーエンド確認応答は、Data Prepper をデータ処理と取り込みに使用する際に堅牢なデータ耐久性を提供します。ただし、いくつかの考慮事項が必要です。

1. エンドツーエンド確認応答は現在、ステートフルな集約と互換性がありません。Data Prepper のメンテナーはこの問題を検討しており、将来のイテレーションでリリースするためのソリューションを模索しています。
2. エンドツーエンド確認応答を活用するには、ソースプラグインが確認応答システムと統合する必要があります。現在、Amazon S3 ソースがこの機能を持つ唯一のソースプラグインです。ただし、この機能は追加のソースを含むように拡張されるよう設計されており、プルベースのソースに適しています。新しい Kafka ソースの開発が進行中であり、これらの確認応答を使用する有望な候補となっています。ユーザーは HTTP などのプッシュベースのソースにこれらの確認応答を適用することにも関心を示しており、メンテナーが検討している領域です。

Amazon S3 にデータが保存されていて OpenSearch に取り込みたい場合は、Data Prepper のエンドツーエンド確認応答機能を試してください。いつものように、Data Prepper コミュニティはフィードバックと貢献を歓迎しています。

この機能の全体的な目標、ユーザーエクスペリエンス、アーキテクチャの詳細については、End-to-end Acknowledgments を参照してください。

この機能に関する フィードバックやコメント を共有してください。