Streamingを中心としたリアルタイムなセンサデータ処理システムの構築方法と、性能検証の進め方、および初期設定における性能測定結果を解説しました。Streaming」、検索エンジン「Elasticsearch」のチューニング方法と、チューニング後の性能測定結果について解説します。初期設定で測定した結果、Kafkaの格納性能は8,026メッセージ/秒、Sparkの処理性能は19,346メッセージ/秒となりました。Kafkaがボトルネックとなり、システム全体のリアルタイム処理は8,026メッセージ/秒となります(図1)。この結果から、初期設定では第2回で解説した目標性能である10,000メッセージ/秒の処理性能を満たすことはできませんでした。図1:初期設定における測定結果KafkaのパラメータチューニングKafkaのProducer(データ送信ライブラリ)はメッセージをバッファリングし、まとめてBroker(キューサーバー)に送信します。そのためBrokerだけでなくProducerのパラメータチューニングも重要になります。ProducerおよびBrokerの仕組みについては第2回の解説を参照してください。Kafkaの格納性能に影響するパラメータProducerの送信性能に影響するパラメータを表1に示します。表1の各パラメータが送信処理のどこに影響するのかを図2に示します。Producerが使用できるメモリ量。この値を増やすとキューイングできるメッセージ数が増え、送信待ちデータが増加した際にも対応できるようになる。ProducerのバッチはBrokerのパーティションに対応し、1Brokerのパーティションに対応する全バッチに格納されたメッセージの合計サイズがこの上限に達するとリクエストが送信される。1つのバッチに格納できる合計メッセージサイズ。Producerはメッセージをバッチという単位でバッファリングする。送信用シングルスレッドのループ待機時間。Requestサイズが上限に達していなくても、この時間が経過したら送信される。Producerからの送信リクエストにBrokerがAck(格納処理結果)を返すタイミング。1即時、2Leader書き込み完了時、3全Followerにレプリケート完了時の3段階で設定可能。同期送信ではProducerからの送信リクエストにBrokerがAckを返してから、非同期送信ではAckの返却を待たずに次の送信を開始する。送信エラーの検知は非同期送信でも可能(Producerのコールバック関数が呼ばれる)。()内はパラメータ名を表す。ProducerのパラメータProducerは表1#2のリクエストサイズ、#3のバッチサイズ、#4の送信待機時間のいずれかが上限に達した時にデータを送信します。今回はパラメータチューニングの効果を明確にするため、送信データ量を調整しやすい#3のバッチサイズのみを変動させて、他のパラメータは固定して測定しました。Brokerの格納性能に影響するパラメータを表2に示します。今回の測定では変更可能なパラメータがないため、すべて固定値としました。1トピック(Kafkaのキュー)を構成するパーティション数。パーティションのレプリカ作成数(メインのパーティションを含むため1個=メインのパーティションのみでレプリカなし)。第2回で説明したデータ保護の要件から、システム障害時にデータを失わないようレプリカを作成する必要がある。()内はパラメータ名を表す。パラメータはコマンドラインからのトピック作成時に設定するKafkaのチューニング結果ここまでに説明した理由から、今回の測定ではProducerのバッチサイズのみを変動させ、他のパラメータは固定して測定を行いました。測定時のパラメータを表3に示します。測定結果を図3に示します。バッチサイズが112KBのときに格納性能が最も高くなることが判明しました。よって、目標性能である10,000メッセージ/秒の格納性能はKafkaのチューニングのみで満たせることになります。Producerのバッチサイズと格納メッセージ数Sparkのパラメータチューニング初期設定での測定結果では、Sparkの処理性能はすでに目標性能(10,000メッセージ/秒)を満たしていましたが、今回はチューニングにより処理性能をどこまで伸ばせるかを確認しました。Sparkの処理性能に影響するパラメータSparkの処理性能に影響するパラメータを表4に示します。Streamingがマイクロバッチ処理を実行する間隔。この処理インターバルごとにKafkaからのメッセージ取得、動作種別の判定、Elasticsearchへの格納が行われる。ワーカノードで起動するエグゼキュータのプロセス数。各エグゼキュータプロセスに割り当てるコア数。各エグゼキュータプロセスに割り当てるメモリ量。ドライバプロセスに割り当てるコア数。ドライバプロセスに割り当てるメモリ量。Sparkが並列処理するタスク数。デフォルト設定では1タスクは1エグゼキュータで処理される。※()内はパラメータ名を表す。パラメータはコマンドラインでのSparkアプリケーション実行時、またはソースコード内で設定するSparkのチューニング結果Kafkaをチューニングした状態でSparkをチューニングして測定を行いました。今回の測定では、表4で示した理由から処理インターバルのみを変動させ、他は固定としました。測定時のパラメータを表5に示します。32個(固定)Sparkの処理インターバルごとの秒間処理メッセージ数を図4に示します。処理インターバルが長くなるほどSparkの処理性能が高くなることが分かります。第2回で説明した通り、今回のシステムでは入力データを5秒以内に処理する必要があるため、処理インターバルも5秒以内である必要があります。よって今回の前提条件では、処理インターバルが5秒のときに最も処理性能が高くなります。結果として、第2回で説明した初期設定における測定時と処理インターバルは変わらず、性能は向上しませんでした。図4:Sparkの処理インターバルと処理メッセージ数Elasticsearchのチューニング図1で示した通り、今回測定したSparkの処理性能はKafkaからデータを取り出してSparkで処理し、Elasticsearchに格納するまでの処理時間です。そのため、Elasticsearchの格納性能をチューニングすることでSparkの処理性能がさらに向上すると考えられます。Elasticsearchのデータ格納処理の流れについては第2回の解説を参照してください。以下に、Elasticsearchのチューニング結果を示します。今回のシステム構成ではSparkのワーカノードにElasticsearchを同居させているため、Sparkクラスタのノード台数=Elasticsearchクラスタのノード台数となります。Elasticsearchの格納性能に影響するパラメータ前回説明したElasticsearchのデータ構造とデータ格納処理の流れから、Elasticsearchの格納性能に影響すると考えられるパラメータを表6に示します。各パラメータが格納処理のどこに影響するのかを図5に示します。SparkがElasticsearchに格納リクエストを発行する間隔。Sparkの処理インターバルがElasticsearchの格納インターバルとなる。1回のBulkリクエストに含められる最大リクエスト数。このリクエスト数を超える場合は複数回に分けてBulkリクエストを行う。リクエストはSparkの各タスクが並列実行するため、このパラメータは1タスクあたりの最大リクエスト数となる。SparkはKafkaから取得したメッセージ(平均で秒間10,112メッセージ)をElasticsearchに格納するため、、1タスクあたりの格納ドキュメント数(=リクエスト数)は秒間10,112ドキュメント×5秒/32タスク=1,580となる。デフォルトのBulkリクエスト数は1,000で、上記のドキュメント数だとリクエストが2回に分割されるだけであり、性能への影響は小さいと考えられる。1回のBulkリクエストの最大サイズ。このサイズを超える場合は、複数回に分けてBulkリクエストを行う。リクエストはSparkの各タスクが並列実行するため、このパラメータは1タスクあたりの最大サイズとなる。#2で説明した通り1タスクあたりの格納ドキュメント数は1,580で1タスクあたりのリクエストサイズは1リクエスト1,580ドキュメント×75byte=約116KBとなる。Elasticsearchへのリクエスト内容をディスク上のトランスログに書き込む間隔。トランスログは永続化前のドキュメントが障害で失われた際の復旧に使用される。システム障害時にデータを失わないためには、トランザクションログをリクエストごとに同期的に書き込む必要がある。トランスログが上限サイズに達するとファイルシステムキャッシュのフラッシュ処理が実施され、メモリ上のデータがディスクに書き込まれる。Kafkaから平均で秒間10,112メッセージを取得するため、測定期間中の格納データ量は10,112メッセージ×300秒×メッセージサイズ75byte=約217MBとなる。インメモリバッファのドキュメントに対してセグメントファイルへの変換を実行する間隔。セグメントファイルに変換されたドキュメントは検索可能。前回のフラッシュからこの設定期間が経過するとフラッシュ処理が実施され、メモリ上のデータがディスクに書き込まれる。インデクスを構成するシャード数。シャードの複製数(メインのシャードは含まない)。複数ノードでクラスタを組む場合はノード間でシャードのレプリケーション処理が行われる。※1SparkとElasticsearchの連携ライブラリ「elasticsearch-spark」の設定であるため、パラメータはコマンドラインでのSparkアプリケーションの実行時に指定するか、Sparkの設定ファイル(spark-defaults.conf)に設定する。()内はパラメータ名を表す※2パラメータはElasticsearchのAPIで設定するか、Elasticsearchの設定ファイルを編集する。()内はパラメータ名を表す図5:ElasticsearchのパラメータElasticsearchのチューニング結果KafkaとSparkがチューニング済みの状態で、Elasticsearchをチューニングして測定を行いました。測定時のパラメータを表7に示します。今回の測定では表6で示した理由から、リフレッシュ間隔以外のパラメータは固定です。1個(固定)Elasticsearchのリフレッシュ間隔を変更した際の秒間格納ドキュメント数の変化を図6に示します。リフレッシュ間隔を60秒に伸ばすことで約5%の性能向上が見られました。上記のSparkとElasticsearchのチューニングでは処理性能が約5%しか改善されませんでした。これは今回のチューニングではレプリカを作成するなどデータの保護を重視したためです。そこで、このデータ保護の要件を無視した場合、どの程度性能が向上するのかを測定しました。Elasticsearchのレプリカ作成を無効にして、トランスログの書き込みを非同期(60秒間隔)に設定しました。測定結果を図7に示します。図7:性能重視のチューニングと処理メッセージ数チューニング結果から、すべてのチューニングを行うことで約14%の性能向上が見られました。Elasticsearchのレプリカの作成を無効にしたことで大きく性能が向上しました。Streaming、Elasticsearchのチューニング方法と、チューニング後の性能測定結果を解説しました。)によってチューニングすべきパラメータは変わります。次回は各OSSに割り当てるマシン台数の調整を行い、システム全体のボトルネックについて考察します。日立製作所OSSソリューションセンタ所属。これまでにストレージシステムとその管理ソフトウェアの開発を手掛けてきた。現在はHadoopやSpark、Kafkaを中心としたビッグデータ関連OSSの導入支援やソリューション開発、およびビッグデータを活用したデータ分析業務を担当している。ITでは、より付加価値の高いコンテンツを会員サービスとして提供しています。ITのWebサイトにログインすることでさまざまな限定特典を入手できるようになります。IT会員サービスの概要とメリットをチェック他にもこの記事が読まれています設計/手法/テスト技術解説データ中心設計(DOA)に基づいたデータモデリング2020/2/20クラウド技術解説Rancherコードリーディング入門(3/3)2020/2/19開発ツールイベントGitHub、エンタープライズ向けのイベントを国内で開催。VeloCloudが選ばれる理由SD-WANとはなにか。Linuxエンジニア育成への思いF5とNGINXの組み合わせで、サービス公開をよりセキュアに。ITがお届けするポッドキャストです。