Kafka message retrieval practice

シーンの課題

メッセージキューの利用において、その分散型の特性上、メッセージ損失やメッセージ再送などの問題は避けられません。

たとえばログ集約のシナリオでは、複数の異種データソースが Kafka にデータを送信し、Spark などの下流コンピューティングエンジンで消費します。一部のログが欠落した場合、メッセージデータの転送方法やデータ構造などが複雑なため、クライアントのログから直接トラブルシューティングを行うことは困難です。

・メッセージ転送の過程では、たとえばコンシューマーが同じデータを重複して消費する場合があります。この場合、メッセージキューから内容に基づいてデータを取得し、メッセージが重複して生産されていないかを確認する必要があります。ただし、メッセージキューはパーティションとコンシューマーオフセットごとに走査するしかなく、柔軟なメッセージ検索を実現できません。

業界の既存のメッセージキュー製品には、メッセージ内容を効果的に検索するツールや手段が不足しており、トラブルシューティングの難易度とコストを大幅に増加させています。

ApsaraMQ for Kafka 検索コンポーネント

検索コンポーネントの概要

ApsaraMQ for Kafka の「検索コンポーネント」は、フルマネージドで高い弾力性を備えたインタラクティブな検索コンポーネントであり、兆規模のメッセージに対するミリ秒レベルの応答性能でメッセージ内容の検索を実現します。業界のメッセージング製品がメッセージ内容の検索をサポートしていないという課題を解決することを目的としています。ApsaraMQ for Kafka の「検索コンポーネント」は、Kafka コネクタを使用して Topic 内のメッセージデータを Tablestore に転送し、Tablestore の複合インデックス機能に基づくメッセージ検索機能を提供します。メッセージのパーティション、オフセット、時間範囲などの条件を 1 つまたは複数組み合わせた検索のほか、メッセージキーとメッセージ値に基づく全文検索もサポートします。

ケース実践

ケースの背景

ある運用保守チームがオンラインクラスターの稼働状況を監視し、プロセスレベルのログを収集して ApsaraMQ for Kafka に取り込むケースを考えます。下流では Realtime Compute for Apache Flink を使用して各プロセスのリソース消費をリアルタイムで計算します。Flink で特定プロセスのログデータが一定期間欠落していることが判明した場合、ApsaraMQ for Kafka の「検索コンポーネント」を使用して、メッセージ値と時間範囲を条件にメッセージデータを検索し、ログが正常に ApsaraMQ for Kafka に配信されたかどうかを確認する必要があります。

たとえば、収集されたログデータは JSON 構造で、1 件のログデータのフォーマットは以下のようになります：

メッセージ検索の有効化

まず、Alibaba Cloud ApsaraMQ for Kafka コンソールにログインし、該当する Topic を選択して、メッセージ検索サービスを有効化します。

メッセージ検索サービスが有効化されると、Tablestore インスタンスが自動的に作成され、メッセージデータが Tablestore に転送された後、インデックスが作成されてメッセージ検索機能が提供されます。各 Topic は Tablestore 内の 1 つのデータテーブルに対応します。ApsaraMQ for Kafka コンソールで、各 Topic のメッセージ検索コンポーネントの詳細を確認できます。

メッセージ検索の実践

メッセージ検索サービスの有効化後、メッセージ内の複数の検索条件を使用してメッセージを検索でき、前述のケースを実現できます。たとえば、時間範囲を指定し、メッセージ値に「PID=276」を含むメッセージを検索します。

2. 返却結果の例

拡張

Tablestore の概要

Tablestore は、飛天プラットフォーム上に構築された構造化データストアであり、数百億件のデータストレージとミリ秒レベルのデータ検索サービスの機能を提供します。ApsaraMQ for Kafka がメッセージを Tablestore に転送した後、Tablestore のネイティブなデータアクセス方法でメッセージを検索できます。Tablestore はより複雑な検索ロジックをサポートし、SQL 構文によるメッセージ検索も可能です。以下に、メッセージ検索の 2 つの方法を紹介します：

複合インデックス検索

1. Tablestore コンソールにログインし、ApsaraMQ for Kafka メッセージデータの転送先インスタンスとデータテーブルにアクセスし、インデックス管理ページで複合インデックス検索を選択します。

たとえば、メッセージ値に「PID=276」または「PID=277」を含むメッセージを検索します。

3. 返却結果

SQL 検索

Tablestore は SQL 構文に基づくメッセージ検索をサポートします。まず、メッセージの転送先データテーブル上に SQL マッピングテーブルを作成する必要があります。

2. Tablestore SQL を使用して、PID=276 のメッセージを検索します。

まとめ

Alibaba Cloud ApsaraMQ for Kafka の「検索コンポーネント」は、メッセージキュー分野で初めてメッセージ内容のインタラクティブな検索に対応したコンポーネントです。Tablestore へのデータ転送に基づくメッセージ検索機能を提供し、キー、値、パーティションなど任意の条件で自由に組み合わせた検索をサポートします。また、キーと値の全文検索もサポートし、開発不要、運用保守不要、高い柔軟性を特長としています。さらに、Tablestore のインデックスや SQL を使用してメッセージを直接検索することもでき、メッセージの存在有無や正確性に関する日常的なトラブルシューティングの速度を大幅に向上させます。