Elasticsearch、Kafka、Cassandra を使用したスト​​リーミング データ センターの構築

過去 1 年間、私は数多くのソフトウェア企業と会い、アプリケーション データ (通常はログとメトリックの形式) の処理方法について議論してきました。こうした議論の中で、時間をかけてデータをまとめるために断片的なツールセットを使用しなければならないことへの不満をよく耳にします。これらのツールには以下が含まれます: - 運用担当者が監視および警告に使用するツール

- 開発者がパフォーマンスを追跡し、問題をトラブルシューティングするためのツール

- ビジネス インテリジェンス (BI) とビジネスがユーザーの行動を分析するために使用する完全に独立したシステム。

これらのツールは異なる視点を使用し、異なるシナリオに適用可能ですが、いずれもデータ ソースとタイプに重点を置いています。多くのソフトウェア チームは、「もっと時間があれば、もっと良いものを作ることができる」と言いますが、率直に言って、世の中には素晴らしいオープン ソース コードがたくさんあるので、ゼロから何かを作る方が理にかなっているかどうかは議論の余地があります。それが私たち Jut がやっていることです。私たちはオープンソースのビッグデータ コンポーネントを使用してストリーミング データ分析システムを構築しました。この投稿では、使用したコンポーネントとそれらを組み合わせる方法について説明します。以下を取り上げます:

- データの取り込み: さまざまな種類のデータストリームを取り込む方法

- データのインデックス作成と保存: 効率的なストレージと統合クエリ

- 連結: システム内のデータフロープロセス

- チューニング: ユーザーが実際に使用できるように、プロセス全体を非常に高速にします。

この記事を読むことで、皆さんが健全かつスケーラブルな方法でシステムを構築し、私たちが遭遇した落とし穴のいくつかを回避するのに役立つことを願っています。

1. データの取り込み

ビジネス分析と監視に関しては、関連するデータの種類、形式、および伝送プロトコルのほとんどは固定されていません。システム内のさまざまなデータ ソースとデータ送信者をサポートできる必要があります。たとえば、データには次のようなものが含まれる場合があります。

- カスタム アプリケーション イベント。

- コンテナレベルのメトリックとログ。

- statsd または収集されたメトリック。

- GitHub や Stripe などのサードパーティからの Webhook イベント。

- アプリケーションまたはサーバーのログ。

- ユーザーの行動。これらは形式や表記が異なりますが、システム内では統一された形式が必要です。どの形式を選択した場合でも、入力データ ストリームを変換する必要があります。

私たちはシンプルで柔軟なデータ形式を選択しました。各レコード (「ポイント」) は一連のキー/値のペアであり、JSON オブジェクトとして便利に表現できます。すべてのポイントには「時間」フィールドがあり、メトリック ポイントには数値の「値」フィールドもあります。他の点は任意の「形状」を持つことができます。フロントエンド HTTPS サーバー (Nginx を実行) はデータを受信し、それを多重分離して、各データ タイプごとにローカルの「コネクタ」プロセス (Node.js を実行) に送信します。これらのプロセスは、受信したデータをシステムの内部形式に変換し、Kafka トピック (信頼性) に公開します。その後、データはインデックス作成や処理に使用できるようになります。上記のデータ型に加えて、コネクタを使用して、受信データをデータ バスに統合するのが最も簡単になるようにすることを検討してください。ここで説明するほどの汎用性や柔軟性は必要ないかもしれませんが、システムがより多くのデータ タイプを取り込むことができ、新しいデータが到着したときに後で再構築する必要がないように、ある程度の柔軟性を設計に組み込むことは常に良いことです。

2データのインデックス作成と保存

このデータはすべてどこかに保存する必要があります。 ***データベース内では、データのニーズが増大するにつれて簡単に拡張できます。また、データベースが分析タイプのクエリをサポートしていれば、さらに便利です。このデータセンターがログとイベントを保存するためだけのものである場合は、Elasticsearch を選択できます。メトリクスだけであれば、時系列データベース (TSDB) を選択できます。しかし、私たち全員がそれに対処する必要があります。最終的には、さまざまな種類のデータを最も効率的に処理できるように、複数のローカル データ ストアを備えたシステムを構築しました。

ElasticSearchはログとイベントを保存します

私たちはイベントデータベースとして Elasticsearch を使用しています。これらのイベントは、発生元に応じてさまざまな「形」を持つことができます。私たちは Elasticsearch API のいくつか、特にクエリ API と集計 API を効果的に使用しています。

CassandraとElasticSearchはメトリクスを保存する

原則として、メトリックは完全に Elasticsearch (またはその他のデータベース) に保存されますが、メトリック データ構造と冗長メトリック データに一致する専用のデータベースを使用する方が効率的です。最善のアプローチは、既存のオープンソースの時系列データベース (TSDB) を使用することです。

これは私たちが最初に使用したもので、オープンソースの TSDB を使用し、バックエンドとして Cassandra を使用しました。このアプローチの課題は、TSDB には Elasticsearch の API とは異なる独自のクエリ API があることです。 API 間の違いにより、イベントとメトリックに対する統一された検索およびクエリ インターフェースを提供することは困難です。そのため、私たちは最終的に、Casandra と Elasticsearch を介してメトリックを保存する独自の TSDB を作成することにしました。

具体的には、時間と値のキーと値のペアを Cassandra に保存し、メタデータを Elasticsearch に保存し、その上にクエリと管理レイヤーを配置します。このようにして、Elasticsearch でイベントとメトリックの検索とクエリを実行できます。ストリーム処理エンジン これで、データとデータベースを取り込むパイプラインができました。フロントエンドアプリケーションを追加してデータを使用する準備はできていますか?いいえ！ Elasticsearch は独自にログとイベントの分析を実行できますが、処理エンジンが必要です。なぜなら：

- リアルタイムデータと履歴データの両方のイベントとメトリックにアクセスするための統一された方法が必要です。

- 状況によっては（監視、アラーム）、発生したときにデータをリアルタイムで処理する必要があります。

- メトリクス！私たちは、指標を見つけて読み出す以上のことをしたいのです

- メトリックは既存のメトリックを改善するように設計されています。

- イベントの場合でも、Elasticsearch API よりも一般的な処理機能が必要です。たとえば、さまざまなソースとデータを結合したり、文字列解析やカスタム集計を実行したりします。ここから、本当に面白くなっていきます。他の人がデータ パイプラインを構築する方法や、Lambda、Kappa などのデータ アーキテクチャについて学ぶには、1 日 (またはそれ以上) かかります。実際に、非常に優れた資料がたくさんあります。早速本題に入りましょう。私たちが実現したのは、リアルタイムのデータストリーミングとバッチコンピューティングの両方をサポートする処理エンジンです。私たちはこれを全面的に支持しており、興味のある方はこことここをご覧いただけます。

ここでは、ストレージや取り込みとは異なり、独自の処理エンジンをゼロから構築しました。これは、他のストリーム処理エンジンがないからではなく、クエリ パフォーマンスを重視しているためです。これについては、次のセクションで別途説明します。具体的には、データ ストリーム処理モデルを実装するストリーム処理エンジンを構築しました。このモデルでは、計算は、集約、ウィンドウ処理、フィルタリング、結合など、入力を出力に変換する操作の有向グラフとして表されます。これにより、クエリとモデルの計算を自然に組み合わせることができ、リアルタイムとバッチに適しており、分散操作に適しています。

もちろん、本当に新しいプロジェクトを構築するつもりがない限り、オープンソースのストリーム処理エンジンを使用することをお勧めします。 Riemann、Spark Streaming、Apache Flink を検討することをお勧めします。

3 クエリと計算

データフローモデル計算に基づくストリーム処理エンジンを使用します。しかし、ユーザーはどのようにクエリを表現し、そのようなデータフローグラフを作成するのでしょうか? 1 つのアプローチは、API または埋め込み DSL を提供することです。インターフェースは、データのクエリとフィルタリング、変換やその他の処理操作の定義を行うメソッドを提供する必要があります。そして最も重要なのは、複数の処理段階を組み合わせてフロー グラフに適用する方法を提供することです。上記の各プロジェクトには独自の API があり、個人の好みはさまざまですが、API の一般的な課題は、SQL アナリストや Excel ユーザーが簡単に使用できないことです。

現時点でこの問題の解決策として考えられるのは、これらの API 上に構築されたツール (単純な Web アプリケーションなど) を通じてユーザーがシステムにアクセスできるようにすることです。

もう 1 つのアプローチは、単純なクエリ言語を提供することです。これが私たち Jut が行っていることです。現在、データ ストリーム用の既存のクエリ言語 (リレーショナル クエリ用の SQL など) がないため、Juttle と呼ばれるデータ ストリーム クエリ言語を作成しました。 Juttle のストリーム グラフ クエリ言語では、基本的に、上の図に示すように、単純な構文を使用して処理パイプラインを宣言できます。

シンプルな構文で、検索、ウィンドウ、結合、集計、グループ化などのプリミティブを備えています。もちろん、フローチャートのデータを処理する前に、データを取得する必要があります。Juttle を使用すると、同じ構文と構造を使用して、イベントやメトリック、リアルタイムや履歴の任意の組み合わせでデータを取得するためのクエリを定義できます。以下はパターンに従った簡単な例です...クエリ |分析する |ビュー (チェーンではパイプ演算子が使用され、構文はシェルに似ていることに注意してください)。 ``` 読み取り -from :1 日前: datatype = 'weblog' | :minute: count() を status_code ごとに減らします | @timechart ```

4 すべてをまとめる: 異常検出の例

これまでは、コンポーネント中心の視点から、コンポーネントとその機能について説明してきましたが、それらがどのように組み合わされるかについてはあまり説明してきませんでした。ここで、視点をデータ中心に切り替えて、リアルタイムクエリと履歴クエリをサポートするために必要な手順を確認しましょう。異常検出アルゴリズムの例を使って説明しましょう。これは良い例です。潜在的な統計モデルをトレーニングするために履歴データを照会し、異常をテストするためにリアルタイム ストリーミング データを照会し、その後、結果をシステムに書き戻して異常を警告する必要があるからです。

ただし、クエリを実行する前に、取り込みプロセスの継続と、受信データがインデックス ストレージに書き込まれる方法をまとめる必要があります。これはインポート サービスによって実行され、時系列データベースへの書き込みタスクが完了し、メトリック データとメタデータが Elasticsearch と Cassandra に保存されます。

ここで、ユーザーが来て異常検出ジョブを開始します。これには、履歴データを読み取り、タスク処理エンジンを介して基礎となるデータベースを直接クエリする必要があります。さまざまなクエリとデータをパフォーマンスのためにさらに最適化したり (後述)、メトリック データベースへの読み取りパスを実装したり (Elasticsearch でメタデータをクエリし、Cassandra でメトリック値を取得し、結果を組み合わせて実際のメトリック ポイントを生成したり) できます。

履歴データは過去の範囲内のデータをカバーし、処理エンジンは履歴データをフロー グラフのリアルタイム データに変換します。これを行うには、処理エンジンがデータをインポート サービスのエントリ ポイントに直接インポートします。データの損失や重複を避けるため、この切り替えは慎重に行う必要があることに注意してください。

この時点で、ライブ データ上で実行されるトレーニング済みの異常検出フロー グラフが完成しました。異常が検出されると、外部システムにアラートを送信するようにします。これは、処理エンジンがデータを外部 HTTP サービスに POST することで実行できます。アラートを送信するだけでなく、内部システムも追跡したいと考えています。言い換えれば、データのストリームをシステムに書き戻すことができるようにしたいのです。概念的には、これは処理エンジン パイプラインを介して取り込みパイプラインにデータを返すことです。

5. チューニング

つまり、データを取り込むための実用的なシステムと、いくつかのデータベースと処理エンジンが揃っています。フロントエンドアプリケーションを追加してデータを分析する準備はできていますか?まだ！実際、可能ですが、問題はクエリのパフォーマンスが依然として非常に遅いことです。クエリが遅いということは、誰も私たちのシステムを使用しなくなることを意味します。

それでは、「統合処理エンジン」の概念をもう一度考えてみましょう。説明したように、これは同じ構造、抽象化、クエリを使用して履歴データまたはリアルタイム データを処理する同じシステムです。パフォーマンス上の課題は、リアルタイム データよりも履歴データのほうがはるかに多いという事実から生じます。たとえば、1 秒あたり 100 万ポイントがシステムに入力され、入力時にリアルタイムでデータを照会できるほど高速なプロセスがあるとします。ここで、同じクエリ セマンティクスを使用して、過去 1 日間のデータをクエリします。

これには、一度に数百億のポイントを処理する必要があります (または、少なくとも、ストレージから読み取る速度に追いつく必要があります)。計算が分散されていると仮定すると、計算ノードを追加することで解決できますが、最悪の場合、これは非効率的でコストがかかります。ここで最適化が重要になります。データ クエリを最適化する方法は多数あります。これらの一部には、クエリ自体の変換が含まれます。たとえば、上流データのフィルターや集計では、クエリのセマンティクスをできるだけ変更しないようにする必要があります。ここで言う最適化とは、データベースにデータのフィルタリングと処理を可能な限り実行させることです。これには次の作業が必要です。

- データベースで処理できるクエリの部分を自動的に識別する

- 対応する部分をターゲットデータベースのクエリ言語に変換する

- バックエンドクエリを実行し、結果をデータフローグラフの正しい場所に挿入します。

6 結論

やったー！もちろん、視覚化レイヤーが必要なければ、これで完了です。システムは API 経由でのみクエリできます。クエリを作成し、データをストリーミングして視覚化し、ダッシュボードを作成するクライアント アプリケーションを構築することもまた難しい問題なので、これについては別の日に説明します。

さて、このデータセンターの建設中に私たちが見たものと聞いたものをまとめてみましょう。

- さまざまなソースから入力データを受け取り、それを統一された形式に変換して、後で使用するために保存する取り込みパイプライン。 (Jut では、これは Kafka 上に構築されます)。

- イベントとメトリックのデータベース。 Jut では、イベントは Elasticsearch を使用し、独自に構築されたメトリック データベースは Cassandra に基づいています。

- 処理エンジン (ラムダ ISH アーキテクチャを使用する場合は 2 つ)。

- システム上でクエリを実行するための API またはクエリ言語。ふう。このシステムの構築は長く興味深い旅でした。独自のシステムを構築したい場合でも、まずは Jut を試してみることができます。役に立つかもしれません。