Apache Sparkとは？

Hadoopよりも高速かつ高機能な分散処理基盤と言われているようです。
複数の分散されたマシンに処理を分散させることで
1台のマシンで処理するよりもずっと速く処理を完了させることができます。
Hadoopとの重要な違いとしては、Hadoopがバッチ処理向きであるのに対し、
Sparkは、データがメモリ上に置かれることもあり、
レイテンシが小さく繰り返し処理に強いという点があります。
そのためインタラクティブに人が操作する処理に向きます。

参考
http://www.publickey1.jp/blog/15/apache_sparksparkntt.html

準備

今回はRaspberry Pi2 B+を3台使用します。
raspmaster・・・マスターとなる１台、ここから各workerマシンを管理します。今回は自身もworkerともしています。
rasp211・・・workerマシン
rasp212・・・workerマシン

Raspberry Piの固定IPアドレスの設定

/etc/network/interfaces
を以下のように書き換えました。

以下を削除

iface eth0 inet dhcp

代わりに以下を記載します。
addressのところはマシン毎に異なります。

auto eth0
iface eth0 inet static
address 192.168.10.212
netmask 255.255.255.0
gateway 192.168.10.1
network 192.168.10.0
broadcast 192.168.10.255

dhcp関連のパッケージは削除しておきます。

sudo apt-get remove dhcpcd5 isc-dhcp-client

/etc/hostsに以下を追加

192.168.10.201  raspmaster
192.168.10.211  rasp211
192.168.10.212  rasp212

これは全てのマシンで必要です。

Java実行環境の確認

Raspbian Jessie(RaspberryPi2で一番メジャーなOS)であれば最初から入っていると思います。

以下のように表示されればOKです。

pi@raspberrypi ~ $ java -version
java version "1.8.0"
Java(TM) SE Runtime Environment (build 1.8.0-b132)
Java HotSpot(TM) Client VM (build 25.0-b70, mixed mode)

spark実行用ユーザーの作成

$ sudo adduser spark

sparkユーザーになっておきます。

su - spark

sparkのダウンロード

wget http://d3kbcqa49mib13.cloudfront.net/spark-1.5.2-bin-hadoop2.6.tgz

hadoop2.6用のURLです。

展開

tar xvfz spark-1.5.2-bin-hadoop2.6.tgz

spark-1.5.2-bin-hadoop2.6ディレクトリが作成されます。

サンプルの動作確認

$ cd spark-1.5.2-bin-hadoop2.6/
$ bin/run-example SparkPi 10

ずらずらと実行のログが画面に表示され

Pi is roughly 3.143584

などのような円周率の計算結果が表示されました。(数値は毎回若干異なるようです)
ラズパイだと１分くらい掛かっていました。

1台だけで動かす場合はここまでで設定終わり。

２台目以降用の設定

sshで各マシン間にアクセスできるようにします。
手順は以下のとおり

raspmaster側のsparkユーザーで実行

$ ssh-keygen

パスフレーズは空にします。

$ ssh-copy-id spark@rasp211
$ ssh-copy-id spark@rasp212
$ ssh-copy-id spark@raspmaster

ssh鍵を各マシンにコピーします。
自分自身のホストになりますが、raspmasterへもこのホスト名でアクセスさせるために設定します。

以下の手順にてraspmasterでworkerマシンを登録します。

$ cp conf/slaves.template conf/slaves
$ vi conf/slaves
> # A Spark Worker will be started on each of the machines listed below.
> raspmaster
> rasp211
> rasp212

localhostを消し、raspmaster,rasp211,rasp212を追加。

各マシンでspark実行の共通設定をします。

$ cp conf/spark-env.sh.template conf/spark-env.sh
$ vi conf/spark-env.sh
> SPARK_MASTER_IP=raspmaster
> SPARK_WORKER_MEMORY=512m

workerマシン(rasp211,rasp212)でも同様にconf/spark-env.shを作成します。(conf/slavesは不要)

Spark分散環境の起動

raspmasterで

$ sbin/start-all.sh

実行

今回は、GroupByTestというサンプルを動かしてみます。
https://github.com/apache/spark/blob/v1.5.2/examples/src/main/scala/org/apache/spark/examples/GroupByTest.scala
まずは１台構成の環境で実行。（--master localが１台の指定です）
末尾の「50」は、対象テストデータの数を指定するパラメータです。大きいほど処理が重くなります。

bin/spark-submit --driver-memory 256m --executor-memory 256m --class org.apache.spark.examples.GroupByTest --master local lib/spark-examples-1.5.2-hadoop2.6.0.jar 50

実行結果

15/12/13 12:21:58 INFO DAGScheduler: Job 1 finished: count at GroupByTest.scala:52, took 45.386944 s
50000

続いてクラスタ構成環境で実行。（--master spark://raspmaster:7077）

$ bin/spark-submit --driver-memory 256m --executor-memory 256m --class org.apache.spark.examples.GroupByTest --master spark://raspmaster:7077 lib/spark-examples-1.5.2-hadoop2.6.0.jar 50

実行結果

15/12/13 12:16:22 INFO DAGScheduler: Job 1 finished: count at GroupByTest.scala:52, took 14.756486 s
50000

45.38秒掛かっていたものが、14.75秒になりました。
ちょうど3倍程度速くなっています。

Spark分散環境の停止