Sparkクラスターを試しにdockerで構築してみた

n番煎じ

概要

• Docker Composeで公式のsparkイメージからクラスターを作成
• Kmeansのサンプルを実行した

Apache Spark

Apache Spark™は、データエンジニアリング、データサイエンス、機械学習をシングルノードマシンやクラスターで実行するための多言語エンジンです。
ApacheSparkホームページの冒頭から引用(訳: DeepL)

分散処理というとHadoopが有名で、それを使って機械学習をば！となるとMahoutってのがありますが、そいつらよりめっちゃ速いらしいです。
個人的にうれしいのは公式がpythonに対応してライブラリを出してくれているところ。機械学習系の話だとやっぱりpythonを使うことが多いので、システムに違和感なく組み込めて良いですね(仕事のコードはそれですごいことになってた)。

環境

• 実行環境: Windows10 Home
• Docker (WSL2で実行)
  • Docker version 20.10.12, build e91ed57
  • Docker Compose version v2.2.3
• 利用したDockerイメージ
  • apache/spark-py (公式HPに載ってたイメージ)
  • なお、ほかにもjupyter等が既に入っているイメージだったり、便利なものが既にあるらしいので、もしかしたらそれらを使った方がスムーズかもしれません

Dockerイメージをビルドする

今回やりたいのは、sparkのクラスタを作り、pysparkで機械学習を実行すること。
なので、docker-compose で apache/spark-pyイメージを引っ張ってきて起動するだけで出来るだろう。
と思っていた時期が僕にもありました。

上記でやろうとするとこうなります。

ModuleNotFoundError: No module named 'numpy'

numpy入って無いんだって(´・ω・)。

今回つくったクラスタで今後遊ぶことも考えると、いろいろパッケージを入れたイメージをビルドしてしまう方がよさそうです。

前置きが長くなりましたが、
まず以下のようなDockerfileを書いてビルドをしました。

my_pyspark.dockerfile

FROM apache/spark-py:v3.2.1
# pythonパッケージインストール
RUN pip install numpy pyarrow scikit-learn ipython

my_pysparkというタグでdockerイメージをビルドします。

docker build -t my_pyspark -f my_pyspark.dockerfile .

Docker Composeでクラスターを起動する

次にdocker-compose.yamlを書きます。
意外と苦戦しました。

version: '3.7'

services:
  master:
    image: 'my_pyspark'
    container_name: 'master'
    user: '0'   # rootユーザーでないとクラスタを起動できない
    tty: true   # コンテナを起動しっぱなしにする
    ports:
      - '8080:8080' # spark UIのデフォルトポート
      - '4040:4040' # spark UIの詳細ページのデフォルトポート
    # commandの意味については後述
    command: 'bash -c "/opt/spark/sbin/start-master.sh && /bin/bash"'
  
  worker1:
    image: 'my_pyspark'
    container_name: 'worker1'
    user: '0'
    tty: true
    ports:
      - '8081:8081' # workerのuiのデフォルトは8081
    depends_on:
      - 'master'
    command: 'bash -c "/opt/spark/sbin/start-worker.sh spark://master:7077 -c 1 -m 2g && /bin/bash"'
  
  worker2:
    image: 'my_pyspark'
    container_name: 'worker2'
    user: '0'
    tty: true
    ports:
      - '8082:8081' # 8081はworker1が使ってるので、ここでは8082
    depends_on:
      - 'master'
    command: 'bash -c "/opt/spark/sbin/start-worker.sh spark://master:7077 -c 1 -m 2g && /bin/bash"'

  worker3:
    image: 'my_pyspark'
    container_name: 'worker3'
    user: '0'
    tty: true
    ports:
      - '8083:8081' # 8081, 8082はworker1,2が使ってるので、ここでは8083
    depends_on:
      - 'master'
    command: 'bash -c "/opt/spark/sbin/start-worker.sh spark://master:7077 -c 1 -m 2g && /bin/bash"'

各サービスのcommandですが、なんかけったいなことになってます。
頭から順に説明します。

bash -c

これはdockerでコマンドを複数実行するときのテクニックです。
-cの引数として渡された "/opt/spark~"が実行されるというわけですね。

start-master.sh と start-worker.sh spark://master:7077 -c 1 -m 2g

sparkのマスターとワーカーをそれぞれ起動するシェルスクリプトを実行しています。
start-worker.sh のspark://master:7077は、マスターとなるサービスを指定しています。
-cと-mはそれぞれ使用するコア数とメモリサイズです。今回はすべてコア1、メモリ2GBとなっています。

/bin/bash

コンテナを起動しっぱなしにする方法ととして、tty:trueにして、コマンドで/bin/bashを実行するというのがあります。
末尾ではこれをやってます。
sparkのマスターやワーカーはバックグラウンドで動くので、コマンドでマスターとワーカーを起動するだけだとdockerはすぐに正常終了してしまいます。

起動

では起動しましょう。
docker-compose up -dで起動します。

起動後にlocalhost:8080にアクセスしてください。
以下のような画面(SparkUI)が表示されるはずです。

ちゃんとワーカーが3つ、つながってますね。

Kmeansのサンプルを実行する

では作成したコンテナで機械学習を試してみましょう。
今回はSparkのMLlibのガイドに載っている、Kmeansのサンプルを実行します。

pyspark-shellの起動

マスターサービスが動いているdockerコンテナに入り、pysparkを起動します。

# コンテナの中に入る
docker-compsoe exec master

# pyspark-shellの実行
$SPARK_HOME/bin/pyspark --master spark://master:7077

起動後にSparkUIをのぞきに行くと、Running Applicationsのところに何か出てきます。

これでpysparkとマスターがつながりました。

Kmeansサンプルの実行

マスターにつながったので、あとはサンプルをコピペするなどして一通り動くことを確認しましょう。
注意として、dataset取得時のload()に渡すパスを、"/opt/spark/data/mllib/sample_kmeans_data.txt"に変更してください。
以下はサンプルのコードをクラスターラベルを取得するところ実行した結果です。

>>> from pyspark.ml.clustering import KMeans
>>> from pyspark.ml.evaluation import ClusteringEvaluator
>>> dataset = spark.read.format("libsvm").load("/opt/spark/data/mllib/sample_kmeans_data.txt")
22/03/13 03:23:52 WARN LibSVMFileFormat: 'numFeatures' option not specified, determining the number of features by going though the input. If you know the number in advance, please specify it via 'numFeatures' option to avoid the extra scan.
>>> kmeans = KMeans().setK(2).setSeed(1)
>>> model = kmeans.fit(dataset)
>>> predictions = model.transform(dataset)
>>> predictions.show()
+-----+--------------------+----------+
|label|            features|prediction|
+-----+--------------------+----------+
|  0.0|           (3,[],[])|         1|
|  1.0|(3,[0,1,2],[0.1,0...|         1|
|  2.0|(3,[0,1,2],[0.2,0...|         1|
|  3.0|(3,[0,1,2],[9.0,9...|         0|
|  4.0|(3,[0,1,2],[9.1,9...|         0|
|  5.0|(3,[0,1,2],[9.2,9...|         0|
+-----+--------------------+----------+

ちゃんとクラスタリングの結果が得られてますね。

最後に

docker-compose でクラスターのお試しをしました。
とりあず動くようなので、今後はこれをつかって遊んでいこうと思います。
気が向いたらそのことも書く予定です。

記事中に何か変なこと書いていた場合は教えてくださいお願いします。