DEIM Forum 2017 B3-2

大規模データ処理フレームワークのストリーミング機能を利用した機械学習処理の評価
一瀬絢衣†

竹房あつ子††

中田 秀基†††

小口正人†† お茶の水女子大学 〒 112-8610 東京都文京区大塚 2-1-1†† 国立情報学研究所 〒 101-8430 東京都千代田区一ツ橋 2-1-2††† 産業技術総合研究所 〒 305-8560 茨城県つくば市梅園 1-1-1
E-mail: †ayae@ogl.is.ocha.ac.jp, ††takefusa@nii.ac.jp , †††hide-nakada@aist.go.jp, ††††oguchi@is.ocha.ac.jp
あらまし各種センサの普及やクラウドコンピューティング技術の習熟に伴い，お年寄りや子供のための安全サービスなどを目的としたライフログの利用が普及してきている．しかし，動画像解析のようなデータ量，計算量の多い処理をクラウドでリアルタイムに行うことは困難である．また，近年ディープラーニング技術の発達で，その高い精度から画像や音声の認識などに広く用いられているが，計算負荷が高いことが問題の一つとなっている．
本研究では，大規模データ処理のための高速かつ汎用性の高いエンジン Apache Spark のストリーミング機能を利用して，ディープラーニングフレームワークの一つである Chainer の機械学習処理を行い，動画像データ解析処理の高速化を図る．本稿ではクライアント側，クラウド側として 2 つの端末間で Apache Kafka を用いてストリーミング処理を行い，クラウド側で機械学習の識別処理を行った．2 つの端末間のネットワーク帯域を変化させ，1 秒あたりに識別処理を行える画像数，1 マイクロバッチの処理にかかる時間を計測し，性能を評価した．結果から，クラウド側の処理が全体のボトルネックとなっていること，低帯域環境においてリアルタイム処理を行うことが可能であることが示された．
キーワード ストリーミング処理，Apache Spark，機械学習，リアルタイム処理

Evaluation of Machine Learning Processing Using a Streaming Functionof a Large-scale Data Processing Framework
Ayae ICHINOSE† , Atsuko TAKEFUSA†† , Hidemoto NAKADA††† , and Masato OGUCHI†† Ochanomizu University
2-1-1 Otsuka, Bunkyouku Tokyo 112-8610 JAPAN†† National Institute of Informatcs 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-8568 JAPANE-mail: †ayae@ogl.is.ocha.ac.jp, ††takefusa@nii.ac.jp , †††hide-nakada@aist.go.jp, ††††oguchi@is.ocha.ac.jp

1. は じ め に解析する手法が提案されている．しかし，動画像解析は連続的に大容量データを転送する必要があり，計算量，データ量の多各種センサの普及やクラウドコンピューティング技術の習熟い処理をクラウドでリアルタイムに行うことは困難である．に伴い，お年寄りや子供のための安全サービスなどを目的とし画像や映像の解析にはディープラーニング技術が広く使われてたライフログの利用が普及してきている．このようなサービスいる．ディープラーニングはニューラルネットワークの中で識では，一般家庭にサーバやストレージを設置して全ての処理を別を行う中間層を多層化したものを用いた機械学習であり，精行うことは困難であるため，センサデータをそのまま送信して度やスピードの向上という点で注目されている．Chainer [1]

クラウドで処理するのが一般的である．センサデータの解析をや Caﬀe [2]，TensorFlow [3] といったディープラーニングのフクラウドで行う研究は多くなされており，Twitter のセンチメレームワークも多く利用されているが，計算負荷が高いことがント分析など小規模かつ大量のデータをクラウド内で効率よく課題の一つとなっている．

本研究では，大規模データ処理のための高速かつ汎用性の高いエンジン Apache Spark(以降，Spark と呼ぶ) [4] のストリーミング機能を利用して，ディープラーニングフレームワークの一つである Chainer を用いて機械学習処理を行い，動画像データ解析処理の高速化を図る．本稿ではクライアント側，クラウド側として 2 つの端末間で Apache Kafka(以降，Kafka と呼図 1 Apache Kafkaぶ) [5] を用いてストリーミング処理を行い，クラウド側で機械学習の識別処理を行った．様々な利用環境を想定して 2 つの端末間のネットワーク帯域を変化させ，1 秒あたりに識別処理をConsumer がデータの到着量を管理することにより Consumer行える画像数を計測し，性能を調査した．実験から，クラウドの最大効率の達成を可能にしている．こういった技術から速度，側の処理が全体のボトルネックとなっていること，低帯域環境耐久性に優れており，リアルタイムのデータパイプラインやスにおいてリアルタイムに識別を行うことが可能であることが示

トリーミングアプリケーションの構築に広く使われている．

された．

2. 3 Chainer
Chainer は，Preferred Networks が開発したディープラーニ

2. 関 連 技 術

ングフレームワークである．Python のライブラリとして提供

2. 1 Apache Spark

されており，制御構造はすべて Python での記述が可能である．

Spark は，カリフォルニア大学バークレー校で開発が開始さ

Chainer の特徴として，柔軟性，直感的，高速の 3 つが掲げられた，大規模データの格納，処理を目的とした分散処理フレーれている．畳込みニューラルネットやリカレントニューラルネッ

ムワークである．同じく分散処理フレームワークである Apache

ト，再帰型ニューラルネットなど様々なネットワークアーキテ

Hadoop で用いられている，MapReduce と呼ばれるバッチ処クチャをシンプルに実装でき，また，GPU にも対応している．

理に特化した処理方法に対し，Spark はデータをメモリに保存

他のフレームワークの多くが一度ニューラルネット全体の構造することで入出力を高速化することにより，処理全体の実行速をメモリ上に展開し，その処理通りに順伝搬，逆伝播を実行す度の向上を図る．このアプローチにより，レスポンスの速さがるというアプローチであるのに対し，「Define-by-Run」方式と必要とされる対話的処理や，データを繰り返し処理する機械学呼ばれる，ネットワーク構築と学習を同時に行う方式を採用し習処理などに適している．ているのも大きな特徴である．動的にネットワークを定義する

Spark は複数のコンポーネントで構成されており，その一つにため手法の自由度が高く，複雑化していくディープラーニング

ストリームデータを処理する Spark Streaming がある．Sparkの開発への対応が可能であると考えられている．また，インス

Streaming は，数秒から数分ほどの短い間隔で繰り返しバッチトールも容易にできることから，広く使われている．

処理を行うマイクロバッチ方式によりストリームデータ処理機

3. 実

験

能を提供する．Spark Streaming は Apache Kafka や Amazon

Kinesis，Twitter からのデータ取得や MQTT のプロトコルに

本研究では，ストリーミング機能を用いた識別処理の性能をも対応しており，それぞれのデータストリームに合ったアプリ調査するため，クライアント，クラウド間のネットワーク帯域ケーションを容易に作成できる．

を変化させた環境下での，1 秒あたりに処理できる画像数を調

2. 2 Apache Kafka

査した．また，処理のリアルタイム性を調査するため，1 マイ

Kafka は，大容量データを高スループット/低レイテンシにクロバッチの処理時間を計測した．

収集，配信することを目的に開発されている分散メッセージ

3. 1 実 験 環 境ングシステムである．複数のサーバ上でクラスタとして実行

本研究では図 2 に示す構成で，クライアント側，クラウド可能であり，Kafka クラスタはトピックと呼ばれるカテゴリに側として 2 つの端末を用いてストリーム処理を行う．クライ

キー，値，タイムスタンプから構成されたレコードのストリー

アント側は Kafka を用いてデータの送信を行う．クラウド側

ムを格納する．メッセージングモデルには送信側がデータの転では Spark で Kafka からデータを受け取り，Python プログラ送を開始する Push 型と，受信側がデータのリクエストを送るムを実行して Chainer を呼び出す．ここで，Spark ではデータことによりデータの転送が開始される Pull 型という 2 つの大は RDD (Resilient Distributed Dataset) という，分散処理をきな概念が存在する．Kafka は図 1 に示す構造により，この 2前提としたオブジェクトのコレクションとして扱われるため，つのメッセージングモデルの概念を一般化している．Producer

データは RDD として読み込み，RDD から Chainer の要求すと Kafka Cluster 間は Push 型のメッセージングモデルであり，る型に変換する．マイクロバッチサイズを指定し，指定した時Kafka Cluster と Consumer 間は Pull 型のメッセージングモ間ごとに区切られたデータに対し Chainer を呼び出して識別処デルである．これにより，Pull 型のデータの処理の分割による理を行った．

処理のスケーラビリティと，Push 型の複数プロセスへのデー実験には 0 から 9 の手書き数字の 28 × 28 画素の画像データタのブロードキャストの両方を可能にしている．Kafka は最大に正解ラベルが与えられているデータセットである MNIST [6]限のスピードでデータを消費することを目標としているため，を用いた．DataProducer から 500,000 枚のデータを流して画図 2 ストリーミング機械学習
表1OSCPU

実験で用いた計算機の性能

Ubuntu 16.04LTS
Intel(R) Xeon(R) CPU W5590 @3.33GHz(8 コア) × 2 ソケット

Memory 8Gbyte
図 4 マイクロバッチサイズを 2 秒に指定した場合の 1 秒あたりの処理画像数

図 3 マイクロバッチサイズを 1 秒に指定した場合の 1 秒あたりの処理画像数

図 5 マイクロバッチサイズを 5 秒に指定した場合の 1 秒あたりの処理画像数

像の識別処理を行った．実験に用いた計算機の性能を表 1 に示す．クライアント側及びクラウド側で同質のノードを用い，その間のネットワーク帯域は 1Gbps となっている．ネットワーク帯域の制御には PSPacer [7] を用いた．

3. 2 スループット計測
クライアント，クラウド間のネットワーク帯域を 10Mbps から 100Mbps まで変化させ，クラウド側でデータを受信し学習処理を行う時間と処理した画像数から，1 秒あたりの処理画像数を計測した．マイクロバッチサイズを 1 秒，2 秒，5 秒に設定して計測した結果をそれぞれ図 3，図 4，図 5，3 つの比較を
図 6 に示す．横軸が端末間のネットワーク帯域を示し，縦軸が1 秒あたりに処理できた画像の枚数を示す．全ての結果において 20Mbps ほどで処理性能が飽和し，ネットワーク帯域が広い
場合においても 1 秒あたりの処理画像数は増加しないことが図 6 マイクロバッチサイズの変化による 1 秒あたりの処理画像数の比較確認できた．つまり，1 秒あたりに送られる画像数が増加して表しているため，1 マイクロバッチのデータ処理にかかる時間もクラウド側の処理がボトルネックとなってしまい，全体のスをタイムスタンプを用いて計測し，平均値を計算して 1 マイクループットが増加しないことがわかった．また，マイクロバッロバッチの処理時間とした．マイクロバッチサイズを 1 秒，2チサイズを 1 秒に設定した場合には 1 秒あたりの処理画像数が秒，5 秒に設定して計測した結果をそれぞれ図 7，図 8，図 9，2,300 枚ほどであるのに対し，5 秒に設定した場合には 2,500 枚

3 つの比較を図 10 に示す．横軸が端末間のネットワーク帯域ほどに増加している．これは，1 マイクロバッチサイズごとにを示し，縦軸が 1 マイクロバッチの処理時間を示す．全ての場データをまとめて識別処理をしているため，Chainer の呼び出合において，10Mbps ではリアルタイムに処理が行われている

しオーバヘッドが削減されたためだと考えられる．

が，20Mbps 以上になると帯域の変化に伴って処理時間も増加

3. 3 リアルタイム性評価していることが確認できた．つまり，受信データが大きくなる

Spark Streaming のマイクロバッチサイズはデータサイズをと処理が追い付かず，リアルタイムに処理できていないことが図 7 マイクロバッチサイズを 1 秒に指定した場合の 1 マイクロバッ
チの処理時間図 10 マイクロバッチサイズの変化による 1 マイクロバッチの処理時間の比較

4. 関 連 研 究
ディープラーニングを用いたストリームデータ解析は近年数多く研究されており，高速に処理するためのアルゴリズムや精度を向上させるアーキテクチャが検討されている [8] [9] [10]．
しかし，これらはストリームデータを一つの計算機で実行することが前提とされている．本研究はセンサ・クラウド間のデータ送信を考慮した全体のスループットを考慮している点で異なる．また，Spark Streaming を利用してストリーム処理を行っているため，Spark の機能を利用した拡張を容易に行うことが可能である．
図 8 マイクロバッチサイズを 2 秒に指定した場合の 1 マイクロバッ
チの処理時間Spark Streaming は 様々な 技 術 に 応 用 さ れ て お り，Miucin
ら [11] は Spark Streaming で実装された解析ツールを提供するツールキットである DINAMITE を提案している．DINA-MITE の解析ツールは高度なデバッグ情報ですべてのメモリアクセスを計測し，プログラマがメモリのボトルネックを特定するのを支援する．Chen ら [12] は，現在広く使われているルールベースのシステムのスピード，スケーラビリティ，フォルトトレランスといった課題への対処として Spark Streaming を利用している．我々は Spark Streaming を用いた機械学習処理により，一般家庭のライフログ解析の効率化を図る．

5. まとめと今後の課題
本研究では大規模データ処理フレームワーク Spark のスト
図 9 マイクロバッチサイズを 5 秒に指定した場合の 1 マイクロバッ
チの処理時間リーミング機能を利用して Chainer を用いた機械学習処理を行う事により，リアルタイムセンサデータ解析処理における性能わかった．Kafka においては Kafka Cluster と Consumer 間は要件を検討した．実験から，低帯域環境でリアルタイムでの画Pull 型のメッセージシステムのデザインであるため，クラウド像識別が可能であることが確認できた．高帯域環境では Kafka側の Chainer の機械学習処理のボトルネックから，データ転送の構造からクラウド側の処理がボトルネックとなってしまってが遅れていることが原因であると考えられる．また全ての結果いることが確認できた．また，高帯域環境でのデータ転送では，で，60Mbps ほどで処理時間がマイクロバッチサイズの 3 倍ほProducer と Kafka Cluster 間の転送がボトルネックとなってどに達し，その後は横ばいになっていることが確認できた．こいることが確認できた．

れは Producer と Kafka Cluster 間のメッセージングシステムがボトルネックとなっているためだと考えられる．

今後の課題としては，クラスタを用いた分散実行により全体のボトルネックとなっているクラウド側の処理の高速化を図る．
また，Producer と Kafka Cluster 間のボトルネックとなっている原因を詳しく解析し，解決する．

謝辞

この成果の一部は，JSPS 科研費 JP16K00177 および国立研
究開発法人新エネルギー・産業技術総合開発機構（ＮＥＤＯ）の委託業務の結果得られたものです．
文

献[1] Tokui, S., Oono, K., Hido, S. and Clayton, J.: Chainer: a
Next-Generation Open Source Framework for Deep Learning, In Proceedings of Workshop on Machine Learning Systems (LearningSys) in The Twenty-ninth Annual Conference on Neural Information Processing Systems (NIPS)(2015). 6 pages.
[2] Jia, Y. et al.: Caﬀe: Convolutional Architecture for
Fast Feature Embedding, arXiv preprint arXiv:1408.5093(2014).
[3] Abadi, M., Agarwal, A., Barham, P., Brevdo, E., Chen,Z., Citro, C., Corrado, G. S., Davis, A., Dean, J., Devin,M., Ghemawat, S., Goodfellow, I., Harp, A., Irving, G., Isard, M., Jia, Y., Jozefowicz, R., Kaiser, L., Kudlur, M.,Levenberg, J., Mané, D., Monga, R., Moore, S., Murray, D., Olah, C., Schuster, M., Shlens, J., Steiner, B.,
Sutskever, I., Talwar, K., Tucker, P., Vanhoucke, V., Vasudevan, V., Viégas, F., Vinyals, O., Warden, P., Wattenberg, M., Wicke, M., Yu, Y. and Zheng, X. TensorFlow: Large-Scale Machine Learning on Heterogeneous
Systems (2015). http://download.tensorflow.org/paper/
whitepaper2015.pdf. pp. 1-19.
[4] Apache Spark, https://spark.apache.org/.
[5] Apache Kafka, https://kafka.apache.org/.
[6] Lecun, Y., Cortes, C. and Burges, C. J. The MNIST
Database of handwritten digits, http://yann.lecun.com/exdb/mnist/.
[7] Takano, R., Kudoh, T., Kodama, Y. and Okazaki, F.: Highresolution Timer-based Packet Pacing Mechanism on Linux
Operating System, IEICE Transactions on Communications, Vol. E94.B, No. 8, pp. 2199–2207 (2011).
[8] Qing, L., Zhaofan, Q., Ting, Y., Tao, M., Yong, R. andJiebo, L.: Action Recognition by Learning Deep MultiGranular Spatio-Temporal Video Representation, Proceedings of the 2016 ACM on International Conference on Multimedia Retrieval, ICMR ’16, New York, NY, USA, ACM,pp. 159–166 (2016).
[9] Ye, H., Wu, Z., Zhao, R.-W., Wang, X., Jiang, Y.-G. andXue, X.: Evaluating Two-Stream CNN for Video Classification, Proceedings of the 5th ACM on International Conference on Multimedia Retrieval, ICMR ’15, New York, NY,USA, ACM, pp. 435–442 (2015).
[10] Read, J., Perez-Cruz, F. and Bifet, A.: Deep Learning in
Partially-labeled Data Streams, Proceedings of the 30th Annual ACM Symposium on Applied Computing, SAC ’15,
New York, NY, USA, ACM, pp. 954–959 (2015).
[11] Miucin, S., Brady, C. and Fedorova, A.: End-to-end Memory Behavior Profiling with DINAMITE, Proceedings of the2016 24th ACM SIGSOFT International Symposium onFoundations of Software Engineering, FSE 2016, New York,NY, USA, ACM, pp. 1042–1046 (2016).
[12] Chen, Y. and Bordbar, B.: DRESS: A Rule Engine on
Spark for Event Stream Processing, Proceedings of the 3rd
IEEE/ACM International Conference on Big Data Computing, Applications and Technologies, BDCAT ’16, NewYork, NY, USA, ACM, pp. 46–51 (2016).