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

n番煎じ

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

# Apache Spark
> Apache Spark™は、データエンジニアリング、データサイエンス、機械学習をシングルノードマシンやクラスターで実行するための多言語エンジンです。
> [ApacheSparkホームページ](https://spark.apache.org/)の冒頭から引用(訳: 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](https://hub.docker.com/r/apache/spark-py/tags) ([公式HP](https://spark.apache.org/downloads.html)に載ってたイメージ)
  * なお、ほかにも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イメージをビルドします。
``` bash
docker build -t my_pyspark -f my_pyspark.dockerfile .
```

# Docker Composeでクラスターを起動する
次にdocker-compose.yamlを書きます。
意外と苦戦しました。

```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)が表示されるはずです。

![ui_master.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/437227/d2e8810c-bd27-d95a-eb96-462efa92d3d0.png)

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

# Kmeansのサンプルを実行する
では作成したコンテナで機械学習を試してみましょう。
今回はSparkのMLlibのガイドに載っている、[Kmeansのサンプル](https://spark.apache.org/docs/latest/ml-clustering.html#k-means)を実行します。
## pyspark-shellの起動
マスターサービスが動いているdockerコンテナに入り、pysparkを起動します。
```bash
# コンテナの中に入る
docker-compsoe exec master

# pyspark-shellの実行
$SPARK_HOME/bin/pyspark --master spark://master:7077
```
起動後にSparkUIをのぞきに行くと、`Running Applications`のところに何か出てきます。

![ui_master_running.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/437227/1779eacc-270b-00db-db29-209363a65425.png)

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

## Kmeansサンプルの実行
マスターにつながったので、あとは[サンプル](https://spark.apache.org/docs/latest/ml-clustering.html#k-means)をコピペするなどして一通り動くことを確認しましょう。
注意として、dataset取得時の`load()`に渡すパスを、`"/opt/spark/data/mllib/sample_kmeans_data.txt"`に変更してください。
以下はサンプルのコードをクラスターラベルを取得するところ実行した結果です。
```bash
>>> 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 でクラスターのお試しをしました。
とりあず動くようなので、今後はこれをつかって遊んでいこうと思います。
気が向いたらそのことも書く予定です。

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