ストームに基づく分散リアルタイム処理アプリケーションの構築に関する予備的研究

最近、空き時間を利用して『Storm』を読み返しました。 Hadoop と慎重に比較すると、前者はリアルタイムのストリーミング データ処理に優れており、後者は MapReduce を介した HDFS に基づくオフライン データ分析と計算に優れていることがわかります。 Hadoop 自体は、リアルタイムのデータ分析や処理には向いていません。両者に共通するのは分散アーキテクチャであり、マスター/スレーブ関係という同様の概念を持っています。

この記事では、Storm クラスターと Zookeeper クラスターをデプロイする方法については詳しく説明しません。代わりに、実際のケースを使用して、Storm を使用してリアルタイム分析とデータ処理を完了する方法を分析したいと思います。

Storm 自体は、Apache がホストするオープン ソースの分散リアルタイム コンピューティング システムです。その前身はTwitter Stormです。 Storm が登場する前は、大量のリアルタイム データ情報を処理する方法のほとんどは、メッセージ キューとワーカー プロセス/スレッドを使用する方法に似ていました。このため、このようなアプリケーションの構築は非常に複雑になります。多くのビジネス ロジックでは、メッセージの送受信、スレッド間の同時実行制御などの問題を考慮する必要があります。ビジネス ロジックはアプリケーション全体のごく一部しか占めない場合があり、ビジネス ロジックを分離することは困難です。しかし、ストームの出現によりこの状況は変化しました。まず、データ ストリーム Stream の抽象概念を抽象化しました。ストリームは、無制限のタプルのシーケンスを参照します。その後、Spouts と Bolts の概念が提案されました。 Spouts は Storm のデータ ソースであり、ストリームの生成を担当します。 Bolts はストリームを入力として受け取り、ストリームを出力として再生成します。また、Bolts は入力ストリームをどのように分割するかを引き続き指定します。 ***Storm は、トポロジの抽象概念を通じて、複数の Spout と Bolt で構成される分散データ処理ネットワークを構成します。 Storm が設計されたとき、Spout と Bolt で構成されるトポロジ ネットワークは、Thrift サービスを通じて意図的にカプセル化されました。このアプローチにより、Storm の Spouts および Bolts コンポーネントを現在の主流の言語で実装できるようになり、フレームワーク全体の互換性とスケーラビリティがさらに向上します。

Storm のトポロジの概念は、Hadoop の MapReduce ジョブの概念と非常に似ています。違いは、Storm トポロジを一度開始すると、強制終了しない限り実行され続けることです。一方、MapReduce ジョブは最終的には終了します。このモデルに基づいて、Storm はリアルタイム データ分析、継続的なコンピューティング、DRPC (分散 RPC) などの処理に非常に適しています。

以下は、Storm を使用してアプリケーションの処理パフォーマンスを向上させる方法を示す、実際のケースに基づいた設計と分析です。

ある通信会社のスパムSMS監視プラットフォームは、各省のスパムSMSの疑いのあるユーザーのスパムSMSコンテンツファイルをリアルタイムでアップロードします。各州は、ファイル内のスパム SMS の内容に基づいて、指定されたセンシティブなキーワードを含むスパム SMS を分析およびフィルタリングし、データベースに保存します。データベースに保存されたスパムテキストメッセージのユーザーは、機密ユーザーとしてリストされ、重要な監視対象となります。結局のところ、これらのスパムテキストメッセージを無差別に送信するのは非常に間違っています。スパム SMS 監視プラットフォームがファイルを生成する速度は驚くべきものです。従来のアプローチでは、各州の各都市に対応する独立したアプリケーションを用意し、機密キーワードを逐次解析してフィルタリングし、保存処理していました。しかし、現状ではプログラム処理のパフォーマンスが効率的ではなく、ファイルのバックログが発生し、時間内に処理して保存できないことがよくあります。

ここで、Storm を通じて上記のアプリケーション シナリオを再編成してみましょう。

まず、次の図に示すように、この場合の Storm クラスターと Zookeeper クラスターの展開について説明します。

Nimbus に対応するホストは 192.168.95.134 で、これは Storm マスター ノードです。他の 2 つのスレーブ ノード スーパーバイザーに対応するホストは、192.168.95.135 (ホスト名: slave1) と 192.168.95.136 (ホスト名: slave2) です。同様に、Zookeeper クラスターも上記のノードにデプロイされます。

Storm は Zookeeper をベースとしているため、Storm クラスターと Zookeeper クラスターは相互に通信します。次に、各ノードの Zookeeper サービスを開始し、次に Storm の Nimbus サービスと Supervisor サービスをそれぞれ開始します。具体的には、Storm インストールの bin ディレクトリでサービスを開始できます。起動コマンドは、storm nimbus > /dev/null 2 ​​> &1 & と storm supervisor > /dev/null 2 ​​> &1 & です。次に、jps を使用して起動の効果を観察します。問題がなければ、Nimbus サービスに対応するホスト上で Storm UI を起動し、対応するサービスを監視します。 Storm インストール ディレクトリの bin ディレクトリでコマンドを入力します: storm ui >/dev/null 2>&1 &。次にブラウザを開き、http://{Nimbus サービスに対応するホスト IP}:8080 と入力します。ここでは、http://192.168.95.134:8080/ と入力します。次の図に示すように、Storm クラスターのデプロイメントを確認します。

Storm のバージョンは 0.9.5 であり、slave1 と slave2 の 2 つのスレーブ ノード (スーパーバイザー) があることがわかります。ワーカーの総数は 8 です (合計スロット)。 Storm クラスターをデプロイし、正常に起動しました。ここで、Storm を使用してこのアプリケーションを書き直し、機密情報をリアルタイムで監視およびフィルタリングできるようにしてみましょう。まず、Storm メソッドのトポロジ構造図を見てみましょう。

SensitiveFileReader-591 と SensitiveFileReader-592 (都市別に分けられたユーザー SMS コレクター) は、Storm の Spouts コンポーネントを表し、データ ソースを示します。ここでは、スパム SMS の疑いのあるユーザーのスパム SMS コンテンツ ファイルをサーバーの指定されたディレクトリから読み取ることを意味します。もちろん、実際のニーズに応じて、Spouts コンポーネントを複数の Spouts に拡張することもできます。

ファイル内の各行の内容を読み取った後、ファイルのコンテンツ コンポーネントが分析されます。ここでは、ファイルの形式コンテンツを分析する役割を持つ SensitiveFileAnalyzer (SMS コンテンツの分解と分析を監視) を指します。

簡単なデモンストレーションのために、ファイル形式を次のように定義します (例のみを記述します)。home_city=591&user_id=5911000&msisdn=10000&sms_content=abc-slave1。各列は & で接続されます。このうち、home_city=591 はスパムテキストメッセージの疑いのあるユーザーの都市コードを表し、591 は福州、592 は厦門を表します。 user_id=5911000 は、スパムテキストメッセージの疑いのあるユーザー ID を表します。 msisdn=10000 は、スパムテキストメッセージの疑いのあるユーザーの携帯電話番号を表します。 sms_content=abc-slave1 はスパムテキストメッセージの内容を表します。 SensitiveFileAnalyzer は、Spouts から「流れる」データを処理するために使用される Storm の Bolt コンポーネントを表します。

***、解析されたデータを企業が指定したセンシティブなキーワードと照合し、フィルタリングしてデータベースに保存します。ここで、フィルタリングされたデータを MySQL データベースに保存します。このタスクを担当するコンポーネントは、SensitiveBatchBolt (機密情報の収集と処理) です。もちろん、これは Storm の Bolt コンポーネントでもあります。さて、上記は Storm の完全なトポロジ構造です。

機密情報の収集、フィルタリング、監視全体のトポロジーについて大まかに理解できたので、次はそれをコードに実装する方法を見てみましょう。まず、次の図に示すように、プロジェクト全体のコード階層を見てみましょう。

まず、定義した機密ユーザー RubbishUsers のデータ構造を見てみましょう。フィルタリングする機密ユーザーのテキスト メッセージに、「racketeer」や「Bad」などの機密キーワードが含まれていると仮定します。具体的なコードは次のとおりです。

ここで、機密情報データ ソース コンポーネント SensitiveFileReader の具体的な実装を見てみましょう。これは、スパムの疑いのあるユーザーのスパム コンテンツ ファイルをサーバーの指定されたディレクトリから読み取り、各データ行を次の Bolt (SensitiveFileAnalyzer) に送信して処理する役割を担います。各ファイルが送信されると、元のファイルは現在のディレクトリ内でサフィックス bak を持つファイルに名前変更されます (もちろん、処理されたファイルを保存するためのバックアップ ディレクトリを再作成することもできます)。 SensitiveFileReader の具体的な実装は次のとおりです。

監視 SMS コンテンツ分解および分析 SensitiveFileAnalyzer のボルト コンポーネントは、データ ソース SensitiveFileReader からデータを受信した後、上記で定義された形式に従ってファイル内の各行のコンテンツを解析し、解析されたコンテンツを次のボルト コンポーネントである SensitiveBatchBolt (機密情報の収集と処理) に送信します。それでは、SensitiveFileAnalyzer Bolt コンポーネントの実装を見てみましょう。

***Bolt コンポーネント SensitiveBatchBolt (機密情報の収集と処理) は、上流の Bolt コンポーネント SensitiveFileAnalyzer から送信されたデータを、企業が指定した機密キーワードと照合します。一致が成功した場合、そのユーザーが監視に重点を置くユーザーであることを意味します。統合管理のために、Hibernate を通じて MySQL データベースに収集します。 ***SensitiveBatchBolt コンポーネントには、収集した機密情報ユーザー データを定期的に出力する監視機能も実装されていることに注意してください。 SensitiveBatchBolt の実装は次のようになります。

これはHibernate経由でMySQLに保存されるため、まずHibernateの設定が与えられます: hibernate.cfg.xml

対応する ORM マッピング構成ファイル rubbish-users.hbm.xml は次のとおりです。

***、Hibernate は引き続き Spring を通じて統合されており、使用されるデータベース接続プールは DBCP です。対応する Spring 構成ファイル jdbc-hibernate-bean.xml の内容は次のとおりです。

これまでに、機密情報をリアルタイムで監視するための Storm コンポーネントの開発はすべて完了しています。それでは、Storm トポロジを完成させましょう。トポロジはローカル トポロジと分散トポロジに分かれているため、ツール クラス StormRunner (トポロジ エグゼキュータ) がカプセル化されています。対応するコードは次のとおりです。

さて、上記のすべてのスパウト/ボルトを「トポロジ」構造につなぎ合わせます。ここでは、デプロイと実行に分散トポロジを使用します。具体的な SensitiveTopology (機密ユーザー監視 Storm トポロジ) コードは次のとおりです。

これまでに、Storm のすべてのコンポーネントが開発されました。ここで、上記のプロジェクトを jar パッケージにパッケージ化し、Storm クラスターで実行します。具体的には、Nimbus に対応する Storm インストール ディレクトリの下の bin ディレクトリに移動し、storm jar + {jar パス} と入力します。

たとえば、次のように入力します: storm jar /home/tj/install/SensitiveTopology.jar newlandframework.storm.topology.SensitiveTopology、次にスパムの疑いのあるユーザーのスパム コンテンツ ファイルを指定されたサーバーの下のディレクトリ (/home/tj/data/591、/home/tj/data/592) に配置し、最後に Storm UI を開いてタスクの起動と実行を確認します (次の図を参照)。

先ほど送信したトポロジ、SensitiveTopology が Storm クラスターに正常に送信されたことがわかります。このとき、SensitiveTopology をクリックすると、以下に示すように、Spouts/Bolts 監視インターフェイスが開きます。

以下を明確に確認できます: Spouts コンポーネント (ユーザー SMS コレクター): SensitiveFileReader591、SensitiveFileReader592 スレッド番号エグゼキューター、タスク送信の出力ステータス。そして、Bolts コンポーネント: SMS コンテンツ分解アナライザー (SensitiveFileAnalyzer) と機密情報の収集および処理 (SensitiveBatchBolt) の動作を監視し、監視を非常に便利にします。

さらに、対応する Supervisor サーバーに対応する Storm インストール ディレクトリの下の logs ディレクトリに移動して、ワーカーの作業ログを表示することもできます。機密情報の監視とフィルタリングの処理を見てみましょう。スクリーンショットは次のとおりです。

SensitiveBatchBolt モジュールの監視スレッドを通じて、現在機密情報を持つ 9 人のユーザーが収集されていることがわかります。機密キーワードを持つこれらのユーザーが MySQL に正常に保存されているかどうかを確認しましょう。

保存結果も 9 であり、ログに出力された数値と一致していることがわかります。そして、スパム テキスト メッセージ sms_content には、確かに「racketeer」や「Bad」などのデリケートなキーワードが含まれています。それはまさに私たちの期待通りです。また、将来的にファイル処理量が増加した場合でも、Spout/Boltの並列度やWorkerの数を調整することで解決できます。もちろん、クラスターの数を水平方向にスケーリングすることでこの問題を解決することもできます。

Apache オープンソース プロジェクトの Storm の Web サイトは http://storm.apache.org/ です。興味のある友達はこまめに注目するといいでしょう。公式 Web サイトには、非常に信頼性の高い技術仕様と、Storm をメッセージ キュー、HDFS、HBase と効果的に統合する方法が記載されています。私の個人的な意見では、アリババは中国でストームの分析と応用を最もうまく行っている企業です。オリジナルの Storm とオープンソースの JStorm を改良しました。興味のある友達は、それにもっと注目することができます。

Storm を使用すると、分散リアルタイム処理アプリケーションを簡単に開発できます。上記のシナリオの設計は、Storm アプリケーションの一例にすぎません。従来のスタンドアロン サーバー アプリケーションと比較して、クラスター化された並列協調コンピューティング処理は、クラウド コンピューティングとビッグ データの時代のトレンドであり、私が今後学ぼうと努める方向でもあります。そこで、私の学習体験をここに書き留めておきます。間違いがあれば、友人たちに批判してもらい、訂正してもらいたいです。