DEIM Forum 2019 J3-5

スケーラブルな分散ストリーム処理基盤の構築と評価
加藤香澄†

竹房あつ子††

中田 秀基†††

小口正人†† お茶の水女子大学〒 112–8610 東京都文京区大塚 2–1–1†† 国立情報学研究所〒 101–8430 東京都千代田区一ツ橋 2–1–2††† 産業技術総合研究所〒 305–8560 茨城県つくば市梅園 1–1–1
E-mail: †{g1320510,oguchi}@is.ocha.ac.jp, ††takefusa@nii.ac.jp, †††hide-nakada@aist.go.jp
あらまし

カメラやセンサ等の普及とクラウド技術の発達により，一般家庭でのライフログの蓄積と利用が拡大している．また，ディープラーニングを用いた画像や音声の認識処理が非常に注目されている．しかし，その処理の重さから一般家庭で収集した動画像データをリアルタイムに識別するのは困難である．我々は，分散プラットフォームとディープラーニングフレームワークを用いた動画像の分散識別処理をストリーミング環境に適応させ，スケーラブルな分散ストリーム処理基盤を構築している．予備実験として，大規模分散プラットフォームの Apache Spark と分散実
行フレームワーク Ray の分散性能を調査し，Ray による高速かつスケーラブルな分散処理が可能であることを示す．
次に，分散メッセージングシステム Apache Kafka と Ray を用いた分散ストリーム処理基盤を構築し，その性能を評価する．実験から，スケーラブルな分散ストリーム処理基盤が構築可能であることを示す．
キーワード 分散処理，ストリーム処理，Apache Kafka

Construction and Evaluation ofa Scalable Distributed Stream Processing Infrastructure
Kasumi KATO† , Atsuko TAKEFUSA†† , Hidemoto NAKADA††† , and Masato OGUCHI†† Ochanomizu University
2-1-1 Otsuka,Bunkyo-Ku, Tokyo 112–8610, Japan†† National Institute of Informatics
2-1-2 Hitotsubashi, Chiyoda-ku, Tokyo, 101-8430, Japan††† National Institute of Advanced Industrial Science and Technology (AIST)
1-1-1 Umezono, Tsukuba, Ibaraki 305–8560, Japan
E-mail: †{g1320510,oguchi}@is.ocha.ac.jp, ††takefusa@nii.ac.jp, †††hide-nakada@aist.go.jp

1. は じ め にる．TensorFlow [1] や Chainer [2] といった多くのディープラー
ニングライブラリが開発されてきた．しかし，ディープラーニ

各種センサの普及とクラウドコンピューティング技術の発達

ングの処理の重さは課題となっている．センサデータを活用すにより，一般家庭におけるライフログの収集と利用が可能とるサービスを運用するには，一般家庭に解析用のサーバやストなった．これらの技術は，お年寄りや子供を見守る安全サービ

レージを設置することは困難であるため、これらのデータはセスや防犯対策・セキュリティといった用途に応用されている．

ンサからクラウドに送られて，クラウドで解析処理される．し加えて，ディープラーニングが画像や音声の識別に広く用いらかし，動画像解析を利用するサービスでは，多くのセンサかられるようになった．ディープラーニングとは，ニューラルネッ

大量のデータが継続してクラウドに送られることが想定され，トワークで識別を行う中間層が多層化された機械学習手法であその解析処理も非常に重いため，リアルタイムに処理するのは

表 1 実験に用いた計算機の性能

OSCPUGPU

Ubuntu 16.04LTS
Intel(R) Xeon(R) CPU W5590 @3.33 GHz
4 core×2 sockets(8 core)
NVIDIA GeForce GTX 980

Memory 45 GiB

図1

想定する大規模分散ストリーム処理基盤
図2 実験環境

非常に困難である．
本研究では，複数センサから送られる大量の動画像データの解析を高効率に行うことを目的とし，スケーラブルな分散ストリーム処理基盤の構築手法を提案する．まず，予備実験として大規模分散プラットフォーム Apache Spark [4]（以降，Sparkと呼ぶ）と分散実行フレームワーク Ray [5] の分散性能の調査を行う．実験から，Ray による高速かつスケーラブルな分散処理が可能であることを示す．次に，分散メッセージングシステ
ムの Apache Kafka [3]（以降，Kafka と呼ぶ）と Ray を用いた分散ストリーム処理基盤を構築し，その性能を示す．実験結果
から，Kafka と Ray を用いた提案分散ストリーム処理基盤が高いスケーラビリティを有することを示す．

図 3 Spark を用いたマスタ・ワーカ処理

2. 分散ストリーム処理基盤
本研究では，図 1 のような大規模ストリームデータ処理基盤を想定している．各一般家庭に設置されたセンサやカメラからクラウドに送信された動画像データは，ストリーム処理基盤によって収集され，分散処理基盤へと渡される．分散処理基盤がデータを受け取ると，データの解析処理がディープラーニングフレームワークを用いて行われ，結果がサービスマネジメントシステムを通してユーザに返される．ストリーム処理基盤として Kafka，分散処理基盤として Spark もしくは Ray の採用を検討している．既発表研究 [6] から，Kafka クラスタにより大量のセンサデータに対するスケーラブルなメッセージングが可能であることを確認している．想定する処理基盤では，Kafkaクラスタが各家庭のセンサから収集した画像データを Spark クラスタもしくは Ray クラスタに送信し，クラスタ内の計算ノードを用いた画像データの識別処理が行われる．動画像データの流量に応じて，スケーラブルに処理可能な処理基盤の構築を目指している．

3. Spark と Ray の分散処理性能調査

れの場合において，1 つ 30KB ほどの大きさの ImageNet デー
タファイルを 10000 個用意し，マスタでファイルを読み出してからワーカで画像データの識別処理が行われ，結果がマスタに返されるまでの時間を計測した．表 1 に実験に用いた計算機の性能を示す．実験に用いるマスタと全ワーカは同質のノードになっている．各ノードは図 2 に示すように 1Gbps のネットワークで接続されている．また，実験では Spark v. 2.3.1，Ray v.

0.5.3，PyTorch v. 0.4.1，TensorFlow v. 1.8.0 を用いた．
3. 1 Spark の分散処理性能調査
図 3 に Spark を用いる場合のマスタ・ワーカ処理の構成を示す．Spark の分散処理は以下のように行われる．
（1）マスタで Python プログラムが実行される．
（2）マスタが ImageNetデータを読み込み，RDD に変換する．
（3）作成した RDD をマスタからワーカに渡す．
（4）ワーカにおいて PyTorch を用いたImageNet データの識別処理が行われる．RDD とは耐障害分散データセットであり，Spark アプリケーション内でデータはRDD に変換されて処理される．RDD と Spark に定義されたメソッドを用いることで Spark は自動的に分散を行う．また，各ノードは Spark Standalone Mode で接続されている．

予備実験として，Spark と Ray の分散処理性能の調査を行う．

Spark と Ray それぞれについてクラスタを構築し，ディープラーニングライブラリ PyTorch [7] とそのバックエンドとしてTensorFlow を用いた画像の識別処理を行う．実験では，データセットとして ImageNet [8] を用いる．Spark と Ray それぞ

ワーカ数を 1 から 5 まで変化させると同時にパーティション数を 8 から 48 まで 8 刻みで変化させた際の実行時間の計測結果を図 4 に示す．パーティション数とは処理するデータ全体の分割数であり，一般的に分割数が多いほどタスクが細かく分か図 4 Spark による分散処理の実行時間

図 5 10Gbps ネットワークでの Spark の実行時間変化

表 2 実験に用いたマスタノードの計算機の性能
インスタンス名 r5.4xlarge

OSUbuntu 16.04LTS
Intel(R) Xeon(R) Platinum 8000 @3.1 GHzCPU

16 coreMemory128 GiB

表 3 実験に用いたワーカノードの計算機の性能
インスタンス名 p2.xlarge

OSCPUUbuntu 16.04LTS
Intel(R) Xeon(R) E5-2686 v4 @ 2.3 GHz
4 coreGPU

NVIDIA K80Memory61 GiB

図 6 Ray を用いたマスタ・ワーカ処理れ分散しやすくなる．図 4 にから，ワーカ数 1-2 間では大きく実行時間が減少しているが，ワーカ数 2 以上では大幅な実行時間の減少が見受けられなかった．加えて，パーティション数を大きくすると性能劣化することが確認できた．リアルタイムス
トリーミング処理では，生成されたセンサデータを順次送信するためパーティションサイズを大きくして高速化を図ることは難しく，パーティションによる性能オーバーヘッドが大きくなることが明らかになった．

Spark の分散挙動確認のため，広帯域ネットワーク環境で図3 と同様の実験を行った．実験には AWS を使用しており，用図 7 Ray による分散処理の実行時間いたマスタノードの性能を表 2，ワーカノードの性能を表 3 に示す．全ノードは最大 10Gbps のネットワークで接続されていて PyTorch を用いた ImageNet データの識別処理が行われる．

る．ワーカ数を 1 から 5，パーティション数を 8 から 48 まで

Spark と同様にワーカ数を 1 から 5 まで変化させて実行時間8 刻みで変化させ，5 回ずつ計測した際の平均値を図 5 に示す．の計測を行った結果を図 7 に示す．Ray には Spark のような

図 5 から，10G ネットワーク環境においても，ワーカ数に応じパーティションの概念がないため，ワーカ 5 台に対してラウンドて処理時間が短くなるが，Spark では計算以外の処理オーバーロビンにタスクを振り分けるようにした．図 7 から，ワーカ数ヘッドが大きいことが確認できた．

3. 2 Ray の分散処理性能調査
図 6 に Ray の場合のマスタ・ワーカ処理の構成を示す．Rayが増加するごとに実行時間が減少している．Ray の ImageNet
データの識別処理は非常にスケーラブルであると言える．

3. 3 分散処理性能の比較の分散処理は以下のように行われる．
（1）マスタで Python プ

表 4 に実験から得られた Spark と Ray の分散処理の実行時間ログラムを実行する．
（2）マスタノード内の Ray ドライバが自全体を以下 2 つのフェーズに分けて示す．
（1）プログラムの開始

身のノードのオブジェクトストアにデータを置き，ローカルスから識別の開始までに要する時間．
（2）識別が開始され，10000ケジューラに連絡する．
（3）ローカルスケジューラはグローバ

個の ImageNet データの識別に要する時間．Spark と Ray のルスケジューラに連絡する．
（4）グローバルスケジューラは各挙動は異なるため，上記 2 つのフェーズはあくまで目安である．

ノードのローカルスケジューラに指令を送る．
（5）ワーカにおい

表 4 に用いている結果は，Spark と Ray それぞれについてワー

表 4 Spark と Ray の処理時間の比較# of workers
Phase
SparkRay

1(1)

5(2)

(1)

(2)

392.873 213.682 229.235 49.897
1.686 226.133

7.123 48.890

図 10 Kafka と Ray を用いた分散ストリーム処理基盤における処理スループットの変化

データを受け取ると Broker は Ray クラスタのワーカノードに起動した Consumer にデータを送信する．
（4）Consumer にお図 8 Ray と Kafka を用いた分散ストリーム処理基盤

いて PyTorch と TensorFlow を用いた画像の識別処理が行われる．実験では，Kafka v. 1.1.0 を用いた．Kafka の処理よりワーカでの処理の負荷が大きいため，Producer 数は 1 に固定する．Ray ワーカおよび Consumer 数を 1，2，4，8，12 と変化させると同時にバッチサイズを 1，5，10，20 と変化させ処理スループットの計測を行った．バッチ化を用いることで一定量のデータの処理をまとめて行い，より高速に大量データを処理できることがある．処理スループットの計測結果を図 10 に示す．縦軸には 1 秒あたりの処理画像数をスループットで示している．図 10 から，バッチサイズ 1-5 間で大幅なスループット向上が見られるが，10-20 間ではさほどスループットが向上していないことがわかる．加えて，全バッチサイズにおいてワー

図 9 Kafka と Ray による分散ストリーム処理の実験構成

カ数増加によりスループットが向上している．以上の結果から，Kafka と Ray を用いたスケーラブルな分散ストリーム処理基盤が構築できることが示せた．

カ数 1 の時と 5 の時のものであり，Spark のパーティション数は 40 となっている．表 4 から，Spark と Ray の識別処理にかかる時間はほぼ変わらない．しかし，Spark は RDD の作成やタスクスケジューリングなどにより，フェーズ（1）の処理時間が非常に長くなっている．Ray の方が Spark より高効率に分散処理が行えることが明らかになった．

4. Ray と Kafka を用いた分散ストリーム処理
実験

5. 関 連 研 究Chen ら [9] は Kafka の Producer-Consumer モデルを動画像のストリーミングデータ取得層に用い，リアルタイム処理フ
レームワーク Spark Streaming を OpenCV と組み合わせデータ処理層とし，HDFS と HBase をストレージ層と最終的な結果表示に用いた．この性能は CPU 使用率に基づき，ワーカのノード数とバッチスライスを調整することにより評価されている．この手法では動画像から簡単な情報を抽出している．さら3 節の結果から，Ray を分散処理基盤に採用し，Ray と Kafkaなる展望として，動的物体の識別や動作の認識のために，CNNを用いた分散ストリーム処理基盤を構築してその性能を調査

や 3D CNN をサポートするモデルによる必要があると述べてする．

いる．我々の手法では，CNN をサポートするディープラーニン

図 8 に Kafka と Ray を用いた提案分散ストリーム処理基盤

グフレームワーク TensorFlow を用いている．また，我々の研の構成を示す．Kafka Broker がクライアントである各家庭のセ究では一般的なデータ処理基盤を用いた高効率な分散ストリーンサ内の Kafka Producer から画像データを受け取り，Ray クム処理に焦点を当てている．

ラスタのワーカ内に起動した Kafka Consumer に渡す．各ワーカにおいて画像データの識別処理が行われる．
図 9 に Kafka と Ray を用いた分散ストリーム処理の実験構成を示す．
（1）Kafka を起動し，Python プログラムを実行する．
（2）画像データが Producer から Broker に渡される．
（3）

6. まとめと今後の予定
Kafka と Ray を用いた分散ストリーム処理基盤を構築し，その性能を調査した．予備実験として，Spark と Ray それぞれについて ImageNet の PyTorch を用いた分散識別処理を行い，その性能を調査した．実験から，Ray による高速かつスケーラブルな分散処理が可能であることを示した．次に，Kafka とRay を用いた分散ストリーム処理基盤を構築し，実験から提案手法が非常に高いスケーラビリティを有することを示した．
今後の課題として，別の環境での提案手法の評価や想定環境である動画像の解析に適応させることを検討している．

謝辞

この成果の一部は，JSPS 科研費 JP16K00177，平成 30 年度国立情報学研究所公募型共同研究，国立研究開発法人新エネ
ルギー・産業技術総合開発機構 (NEDO) の委託業務および JST

CREST JPMJCR1503 の支援を受けて得られたものです.
文献[1] M. Abadi, A. Agarwal, P. Barham, E. Brevdo, Z. Chen,C. Citro, G. S. Corrado, A. Davis, J. Dean, M. Devin,
S. Ghemawat, I. Goodfellow, A. Harp, G. Irving, M. Isard, Y. Jia, R. Jozefowicz, L. Kaiser, M. Kudlur, J.
Levenberg, D. Mane, R. Monga, S. Moore, D. Murray, C. Olah, M. Schuster, J. Shlens, B. Steiner, I.
Sutskever, K. Talwar, P. Tucker, V. Vanhoucke, V. Vasudevan, F. Viegas, O. Vinyals, P. Warden, M. Wattenberg,
M. Wicke, Y. Yu, and X. Zheng,
TensorFlow: Largescale machine learning on heterogeneous systems, 2015,
http://download.tensorflow.org/paper/whitepaper2015.pdf.
pp. 1- 19. [Online]. Available: http://tensorflow.org/[2] S. Tokui, K. Oono, S. Hido, and J. Clayton,
Chainer:a next-generation open source framework for deep learning, in In Proceedings of Workshop on Machine Learning
Systems (LearningSys) in The Twenty-ninth Annual Conference on Neural Information Processing Systems (NIPS),
2015, 6 pages.
[3]
Apache kafka, https://kafka.apache.org/[4]
Apache Spark, https://spark.apache.org/.
[5] P. Moritz, R. Nishihara, S. Wang, A. Tumanov, R. Liax,
E. Liang, W. Paul, M. I. Jordan, I. Stoica, Ray: A Distributed Framework for Emerging AI Applications, 2017.
Available: http://ray.readthedocs.io/en/latest/index.html[6] 一瀬絢衣，竹房あつ子，中田秀基，小口正人, Kafka を利用したリアルタイム動画像解析フレームワークのレプリケーションによる性能変化の調査, 第 10 回データ工学と情報マネジメン
トに関するフォーラム (DEIM2018), I3-3, 018.
[7] A. Paszke, S. Gross, S. Chintala, G. Chanan, Pytorh: Tensors and dynamic neural networks in python with strong gpu
acceleration, 2017. Available: https://pytorch.org/[8] J. Deng, W. Dong, R. Socher, K. Li, L. Fei-Fei, ImageNet:
A large-scale hierarchical image database. IEEE Conference on Computer Vision and Pattern Recognition, 2009.
Available: http://www.image-net.org/[9] H. Chen, F. Luo, L. Zhao, and Y. Li, Design and Implementation of Real-Time Video Big Data Platform Based onSpark Streaming, International Conference on Computer
Science and Application Engineering (CSAE), 2017.