Kafka+Spark Streaming+Elasticserachによるシステム構築と検証の進め方

はじめに

前回はSpark Streamingの概要と検証シナリオ、および構築するシステムの概要を解説しました。今回はシステムの詳細構成と検証の進め方、および初期設定における性能測定結果について解説します。

この検証ではメッセージキューのKafka、ストリームデータ処理のSpark Streaming、検索エンジンのElasticsearchを組み合わせたリアルタイムのセンサデータ処理システムを構築しています。今回はKafkaとElasticsearchの詳細なアーキテクチャやKafkaとSparkの接続時の注意点も解説します。

システムの詳細構成

マシン構成とマシンスペック

評価に向けたマシンの初期構成を図1に示します。本システムは以下のノードから構成されます。

• センサデータを収集してKafkaに送信する収集・配信ノード
• Kafkaクラスタを構成してメッセージの受け渡しを行うキューとして動作するKafkaノード
• SparkクラスタにアプリケーションをデプロイするSpark Clientノード
• Sparkクラスタのリソース管理をするSpark Masterノード
• Sparkアプリケーションの実行とデータ蓄積を行うSpark Worker＋Elasticsearchノード
• 蓄積したデータを可視化するKibanaノード

今回は仮想化環境を利用して性能評価を実施しました。初期構成のマシンスペックを表1に示します。

測定環境

また、今回の測定は仮想化環境上で実施したため、物理環境とはディスク性能やネットワーク帯域が異なります。検証前に測定したディスク性能とネットワーク帯域を表2に示します。

プログラムの作成とデータセットの準備

自作したプログラム

今回の検証では、下記のプログラムを自作しました。

• データ配信プログラム（BasicKafkaProducer.javaを参照）
  時系列のセンサデータが記述されたテキストファイルを読み込み、Kafkaへ擬似的にストリーム配信を行うプログラム（Javaで開発）
• 動作判定プログラム（HumanActivityClassifier.scalaを参照）
  Kafkaからセンサデータを読み出し、センサデータから動作種別を判定して判定結果をElasticsearchへ格納するプログラム（Scalaで開発したSparkアプリケーション）

なお、Kafkaからのデータ収集とElasticsearchへの格納はSpark用のライブラリを使用します。また、動作種別の判定には事前に学習済みの機械学習モデルを使用します。このモデルについては次節で説明します。

上記プログラムの位置づけを図2に示します。

検証用データセットとデータ変換内容

検証で使用するデータセットとシステム処理中のデータ変換内容は下記のようになります。

(1)学習用データ

本システムでは、Sparkの機械学習コンポーネントMLlibを使用して、事前にセンサデータから動作種別を判別するモデル（ロジスティック回帰モデル）を作成しています。この学習用データには表3に示すUCIリポジトリのオープンデータ（Human Activity Recognition Using Smartphones Data Set）を使用しました。この動作種別モデルは前述した動作判定プログラムが使用します。

(2)配信データ

測定時には表3の評価用データを使用します。前述したデータ配信プログラムがテキストファイルから評価用データを読み込み、時刻と端末IDを付与してJSON形式のデータに変換してKafkaへ配信します。配信データの詳細を表4に示します。

(3)出力データ

Sparkアプリケーションは、表4の配信データに含まれるセンサデータから動作種別を判定します。判定後の動作種別は表3の出力値で示した以下の6種類です。

1. 歩いている
2. 階段を上っている
3. 階段を下っている
4. 座っている
5. 立っている
6. 寝ている

また、SparkアプリケーションはUNIX time表記の時刻を文字列表記に変換します。Sparkアプリケーションの変換結果はElasticsearchに格納されます。変換結果の例（約75byteのJSON形式データ）を以下に示します。

{
      "time": "2016/03/31 12:00:00.000",
       "user": "Test User",
       "activity": "RUNNING"
}

システムの処理の流れ

本システムの処理の流れを以下に示します。

1. 収集サーバ上のデータ配信プログラムはテキストファイルに記述されたセンサデータを一定間隔で読み込み、疑似的なストリーミングデータとしてKafkaに送信する
2. Kafkaは処理データ量の増加に対応するため、収集サーバから受信したデータをキューイングする
3. Sparkアプリケーションは一定間隔でKafkaからデータを読み出し、学習済みの動作種別モデルを用いてセンサデータから動作種別を判定してElasticsearchに格納する
4. KibanaはElasticsearchに格納された動作種別の時系列データを可視化する

以上のシステムの処理の流れを図3に示します。

検証の進め方

今回の検証では、まずデフォルトのパラメータで設定した各OSSを用いて、単位時間当たりの処理メッセージ数（データ量）を測定します。その後、各OSSのパラメータチューニングとシステム構成の変更を行い、性能がどこまで改善するかを検証します。

測定範囲

性能の測定範囲を図4に示します。今回のシステムでは、配信サーバからKafkaにデータを格納するまでの処理とKafkaからデータを取り出してSparkで処理し、Elasticsearchに格納するまでの処理がそれぞれ一連の処理となります。測定項目を表5に示します。

目標性能と前提条件

第1回では、以下をサービスの前提条件として設定しました。

• サービスの同時利用者数は10,000人とする
• 入力データは最低1秒間隔で音楽配信サービスに通知される
• 動作の変化を判断した場合は5秒以内に選曲を行う

今回は、この前提条件から、以下の目標性能を設定します。

• Kafkaの格納性能とSparkの処理性能は共に10,000メッセージ/秒以上
• Sparkの処理インターバルは5秒以内

また、本システムではモバイル端末のストレージ容量を節約するため、送信済みのデータはモバイル端末に残さない前提とします。そのため、システム障害時にはモバイル端末から受信したデータを失わないようにする必要があります。そこで、以下のようなデータ保護に関する要件を追加します。

• KafkaおよびElasticsearchではデータのレプリカを作成する
• Elasticsearchではトランザクションログを同期的に書き込む

上記の要件にあるデータのレプリカ作成とElasticsearchのトランザクションログの詳細については後述します。

測定方法と性能の算出方法

今回の測定は、Kafkaへのメッセージ格納とSparkによるメッセージ取得・動作判定・格納処理を並列で実行した状態で行いました。Kafkaに300秒間メッセージを格納し続け、SparkはKafkaからメッセージを5秒間隔で取得し、動作判定とElasticsearchへの格納を行います。この300秒間の処理における秒間処理メッセージ数を測定しています。

KafkaはProducerからBrokerに書き込みした秒間メッセージ数を使用します。SparkはKafkaが格納したメッセージを1インターバル（5秒）のうち何秒で処理できたかを元に秒間処理メッセージ数を算出します。例えばKafkaに秒間10,000メッセージが格納され、それをSparkが1インターバル（5秒）のうち2.5秒ですべてを処理した場合、50,000/2.5=秒間20,000メッセージを処理したと計算します。

選択したOSSの詳細

今回の性能測定では、SparkのほかにKafkaとElasticsearchの性能が影響します。そのため、ここで改めてKafkaとElasticsearchの詳細を説明します。

Kafka

KafkaはPub/Subメッセージングモデルを採用した分散メッセージキューであり、スケーラビリティに優れた構成となっています（図5）。

Kafkaは複数台のBrokerノードでクラスタを構成し、クラスタ上にTopicと呼ばれるキューを作成します。 書き込み側は入力メッセージをProducerという書き込み用ライブラリを通じてBrokerクラスタ上のTopicに書き込み、読み出し側はConsumerという読み出し用ライブラリを通じてTopicからメッセージを取り出します。

Kafkaの特徴を表6に、詳細なアーキテクチャを図6に示します。

Kafkaは仮想的な1つのキュー（Topic）を複数のノード（Broker）上に分散配置したパーティション（Partition）で構成します。このパーティション単位でデータを書き込み/読み込みして1つのキュー（Topic）に並列書き込み/読み出しを実現します。パーティション内のメッセージは一定期間が経過した後で自動的に削除されます。また、パーティションの容量を指定して容量を超えた分のメッセージを自動的に削除することも可能です。

書き込み側のアプリケーションはProducerを使用してメッセージを送信します。メッセージはランダムにTopicのどれか1つのパーティションに書き込まれます。Producerの仕組みについては後述します。

読み出し側のアプリケーションは1つ以上のConsumerを使用してConsumerグループを構成し、メッセージを並列に読み出します。Topicの各パーティションはConsumerグループ内の特定の1Consumerのみが読み出します。これによりTopicのメッセージを並列かつ（Consumerグループ内では）重複なく読み出すことができます。

また、各Consumerがメッセージをどこまで読み出したかはConsumer側で管理し、Broker側では排他制御を行いません。そのため、Consumer数が増加してもBroker側の負担は少なくて済みます。

Kafkaはクラスタ内のBroker間でパーティションのレプリカを作成します（図7）。レプリカの作成数は指定可能です。レプリカはLeader/Follower型と呼ばれ、読み書きできるのはLeaderのみです。メッセージはLeader/Follower共にOSページキャッシュに書き込まれるため、永続化の保証はありません（定期的にディスクへ書き込まれます）。BrokerはProducerがパーティションに書き込むときにAckを返します。このAckの返却タイミングは即時、Leaderの書き込み完了時、全Followerのレプリケート完了時のいずれかを指定できます。

Producerの仕組みを図8に示します。ユーザアプリケーションはProducerのAPIを通じて送信したいメッセージを登録します。Producerは登録されたメッセージをBatchという単位でバッファリングします。Batchはパーティション単位でキューイングされ、各キューの先頭のBatchがBroker単位でまとめて送信されます（これをリクエストと呼びます）。Brokerは受信したリクエストに含まれる各Batch内のメッセージを対応するパーティションに格納します。

Elasticsearch

Elasticsearchは全文検索エンジンです。Elasticsearchのデータ構造とデータ格納処理の流れを解説します。

(a)Elasticsearchのデータ構造

Elasticsearchのデータ構造を図9に示します。Elasticsearchは複数台のノードでクラスタを組み、データを分散して保持できます。またIndex（RDBMSにおけるDatabaseに相当）を各ノードに分散させた複数のシャードで構成します。シャードは耐障害性を確保するためにレプリカを作成できます（デフォルトでは1個）。Index内には複数のType（RDBMSにおけるTableに相当）を作成でき、Typeには複数のドキュメント（RDBMSにおけるレコード（Tableの一行）に相当）を格納します。

今回構築したシステムでは、Sparkで動作種別を判定したメッセージをElasticsearchにドキュメントとして格納しています。

(b)Elasticsearchのデータ格納処理の流れ

Elasticsearchのデータ格納処理の流れを以下に示します。

（1）格納リクエスト

Sparkは動作種別の判定結果をElasticsearchに格納するため、処理インターバルごとに格納リクエストを発行します。これにはElastic社が提供するSpark用のライブラリを使用します。

このライブラリでは格納リクエストにBulkリクエストを使用します。Bulkリクエストには1回のリクエストに複数のリクエストを含ませることができ、これを利用して複数のドキュメントを1回のリクエストにまとめて格納します。なお、格納リクエストのプロトコルはHTTP POSTです。

（2）インメモリバッファに格納
ElasticsearchがIndexリクエストで受け取ったドキュメントは、まずインメモリバッファに書き込まれます。

（3）トランスログを書き込み
Elasticsearchはリクエスト内容をディスク上のトランスログ（トランザクションログ）に書き込みます。デフォルト設定ではリクエストごとに同期書き込みを行います。このトランスログは永続化前のドキュメントが障害により失われた際の復旧に使用されます。

（4）リフレッシュ（ソフトコミット）
Elasticsearchでは定期的（デフォルトでは1秒間隔）にリフレッシュ処理が行われ、インメモリバッファ上のドキュメントが検索可能となります。これは擬似リアルタイム検索を実現するための仕組みです。リフレッシュ処理が呼ばれると、インメモリバッファ上のドキュメントはまとめてセグメントという固まりに変換され、ファイルシステムキャッシュ上に配置されます。

（5）フラッシュ（ハードコミット）
Elasticsearchでは以下のいずれかのタイミングでフラッシュ処理が行われ、リフレッシュ処理とファイルシステムキャッシュ上のセグメントのディスク書き込みが行われます。

• トランスログのサイズが上限（デフォルトでは512MB）に達した
• 前回のフラッシュから一定時間（デフォルトでは30分）が経過した
• 前回のフラッシュから一定回数（デフォルトでは無制限）の操作（リクエストなど）が行われた

フラッシュ処理が完了するとメモリ上のドキュメントはすべて永続化されるため、トランスログは不要となり消去されます。

以上のデータ格納処理の流れを図10に示します。

測定の初期設定と測定結果

測定の初期設定

今回の評価では、基本的に各OSSで設定可能なパラメータはデフォルト値を利用します。デフォルト値がなく設定が必要なパラメータと、デフォルト値から変更したパラメータを表7に示します。

※1 ドライバプログラムはSparkアプリケーションの実行中にワーカノードに常駐してアプリケーション全体のタスク実行を管理する
※2 エグゼキュータはSparkアプリケーションの実行中にワーカノードに常駐してタスクを実行する

表7でKafkaのパーティション数を32個に設定した理由について解説しましょう。まず、SparkがKafkaからデータを取得する方式には2種類（Spark Streaming + Kafka Integration Guide）があります（表8）。

今回の検証では、レシーバタスクを使用しない方式を採用しました。この方式ではKafkaのパーティション数と同数のSparkタスクが自動生成されます。Sparkでは1タスクを1コアで処理するため、Sparkに割り当てられたコア数よりタスク数が少ない場合、一部のコアは使用されないことになります。

表7で説明した通り、検証ではSparkがワーカノード4台（4エグゼキュータ）を使用し、各ワーカノードのCPUは8コアであるため、Sparkが処理に使用できるコア数は4ワーカノード×8コア＝32コアとなります。Sparkのタスク数をコア数と同数の32タスクにするため、今回の検証ではKafkaのパーティション数を32個としました。

初期設定における測定結果

表7の初期設定で測定した結果、Kafkaには1秒間で平均8,026メッセージが格納され、それをSparkが1インターバル5秒のうち平均2.07秒ですべて処理しました。Kafkaの格納性能は8,026メッセージ/秒、Sparkの処理性能は8,026×5/2.07=19,346メッセージ/秒になります。

よってKafkaがボトルネックとなり、システム全体でリアルタイムに処理できるのは8,026メッセージ/秒となります（図11）。デフォルト設定では、目標性能である10,000メッセージ/秒の処理性能を満たすことはできませんでした。

おわりに

今回はシステムの詳細構成から、初期設定における検証結果までを解説しました。次回は、システムのパラメータチューニングを行い、性能がどこまで改善したのかについて解説します。