Hadoop不要！Sparkクラスタを手軽に構築

こんにちは、平野です。

SparkはHadoopと一緒に語られることが多いのでHadoopと一緒でないと分散処理環境が使えないと思っている方もいるかもしれませんが、Sparkだけで分散処理のクラスタを作る機能があり、構築も非常に簡単ですので、その方法を紹介します。

以下、Sparkをインストールした場所をSPARK_HOMEとします。 なおインストールの手順は こちら の記事を参照してください。

手順

まずは、クラスタのマスタープロセスを立ち上げます。 クラスタを管理するコマンド一式は＄SPARK_HOME/sbin/ディレクトリの中に含まれています。

$ cd $SPARK_HOME
$ sbin/start-master.sh
starting org.apache.spark.deploy.master.Master, logging to /Users/hirano.shigetoshi/Spark/spark-2.3.2-bin-hadoop2.7/logs/spark-hirano.shigetoshi-org.apache.spark.deploy.master.Master-1-HL00341.local.out

あっさりと立ち上がるかと思います。 立ち上がったこのSparkのクラスターは以下のアドレスからWebUIで様子を見ることができます。

http://<Master-node-IP>:8080

私は手元のMacで行なったので http://localhost:8080 で大丈夫でした。

続いてスレーブのプロセスを立ち上げます。 スレーブは、ホスト単位での登録というイメージです。 スレーブプロセスを立ち上げたホストは、「このホストはSparkのスレーブとして使う」という位置付けなので、基本的に全リソースをSparkに提供します（もちろん設定で変更可能です）。

ですので、基本的にはスレーブはマスターと分けるのが大原則ですが、ここではひとまずマスターが動いているのと同じホストにスレーブとしても働いてもらいます。

起動はsbin/start-slave.shに、先ほどWebUIで見たクラスタのURLを指定しれあげればOKです。

$ sbin/start-slave.sh spark://HL00341.local:7077
starting org.apache.spark.deploy.worker.Worker, logging to /Users/hirano.shigetoshi/Spark/spark-2.3.2-bin-hadoop2.7/logs/spark-hirano.shigetoshi-org.apache.spark.deploy.worker.Worker-1-HL00341.local.out

spark://HL00341.local:7077 がSparkクラスターのIDで、先ほどのWebUIに表示されているものをそのまま貼り付ければ大丈夫です。 （"HL00341"は私のMacの名前です。） 先ほどのWebUIを更新すると、ワーカーノードが増えていることが確認できるかと思います。

最後に、使用するクラスタURLを指定してpysparkを立ち上げます。

$ bin/pyspark --master spark://HL00341.local:7077

masterを特に指定しないとローカルでの動作になりますが、masterを指定することで、Sparkの処理はそのクラスタで実行されます。

一連の流れとしては以上で完了です。
マスターを立ち上げて、そのクラスタのURLを与えてスレーブを立ち上げる。 pyspark自体もそのクラスタURLを指定して立ち上げる。 かなりシンプルで分かりやすいので、すぐにでも使えそうです。

クラスタの停止

スレーブを止めるなら、そのホストで以下を実行するだけです。

$ sbin/stop-slave.sh

マスターを停止させる際には、予めスレーブを全て停止させた後

$ sbin/stop-master.sh

です。 簡単で助かります。

指定クラスタで動いているかの確認

確かに今起動したクラスタで処理が行われているかの確認のために、ここでは逆説的に、スレーブが一つもない状態では処理が動かないということを確認したいと思います。

上記の通りスレーブをストップさせ、ワーカーノードが０であることを確認します。 その上で、pyspark上で

sc.parallelize([1,2,3]).collect()

など、すぐに終わるコードを実行します。

なんという事もない仕事ですが、ワーカーが一人もいないのでいつまでたっても結果が返ってこないことが確認できるかと思います。

SlaveがSparkに使うリソースについて

上でも述べたように、特に何もオプションを指定せずにスレーブプロセスを立ち上げた場合、そのホストはほぼ全リソースをSparkに明け渡すような設定になります。 具体的には以下のようなリソースを提供します。

• CPU Core数
  • そのホストの全Cores数
• メモリ
  • そのホストの全メモリ - 1GB

ですので、実際にいくつかのスレーブホストをSpark専用にぶら下げるのであれば、特に何もオプションを与えないままで良いかと思いますが、テストで動かしてみる場合には、

$ bin/pyspark --master spark://HL00341.local:7077 -c 1 -m 2G

などの設定で起動するのが良さそうです。

複数のスレーブホストを使用する場合

上ではマスターと同一のホストでスレーブを立ち上げてしまいましたが、単純に他のホストにログインして、同様にスレーブプロセスを立ちあげれば、簡単に分散処理環境が出来上がります。

またスレーブの数が増えてきた場合には、複数のホストで一発起動ということもできます。 $SPARK_HOME/conf/slavesというファイルを作り（slaves.templateから生成）、そこにスレーブプロセスを起動するホストを列記して、マスタープロセスの走っているホストで

$ sbin/start-slaves.sh

を実行します。 これでconf/slaveに列挙した各ノードでスレーブプロセスが立ち上がります。

ここではクラスタのURLを指定していませんが、どうやらマスターのホストで実行した場合は、デフォルトで良きに計らってくれるようです。 具体的に指定したい場合には、conf/spark-env.shに、クラスタURLを分解して

export SPARK_MASTER_HOST=HL00341.local
export SPARK_MASTER_PORT=7077

のように指定します。

またクラスタURLだけでなく、Core数などのオプションもconf/spark-env.shに書きます。 1Core、2GBであれば

export SPARK_WORKER_CORES=1
export SPARK_WORKER_MEMORY=2G

という設定をファイルに追記すればOKです。 その他詳細については こちら を参照してください。

また、startと同様にstopも、sbin/stop-slaves.shが使えます。 これはconf/slavesを参照することも含めて同様です。

masterの指定について

Spark独自の分散環境の指定に bin/pyspark --master <spark-cluster-url> と起動しましたが、Sparkはこの--masterにクラスタを指定することで、様々な分散環境を利用することができます。
例えば既存のHadoopクラスタで実行する場合は--master yarnなどと指定します。 他にもKubernetes上での実行などもできるようですので、そちらもいつか試してみたいです（Kubernetesまだ全然理解していないのですが・・・）。
このように、masterの指定を変えるだけで対象のクラスタを変えることができるので、Sparkは非常にフットワークが軽く動作させることができます。

なお、単にbin/pysparkとだけした場合は、実際はbin/pyspark --master local[*]が実行され、ローカルモードでの起動になります。 [*]は全てのCoreを使用するという意味です。

まとめ

Spark独自の分散環境を作成する手順をご紹介しました。

SparkはHadoopなしでも非常に簡単に分散環境を構築できるので、使用するリソースを制限すれば既存のオンプレ環境を間借りする形でも始めやすいかと思います。

興味を持った方は試してみて頂ければと思います。