手を動かす Spark MLlib & Word2Vec Part 1 （spark-ec2 でクラスタを構築するまで）

このシリーズについて

機械学習系のツールを全然使ったことがなかったので、勉強のためになにか1つ選んで、実際に手を動かしてみることにしました。マシンを並べて負荷分散することを想定して、まずは Spark MLlib を選びました。

このシリーズでは、Amazon EC2 上に構築した Spark Cluster (Standalone Mode) で、Wikipedia のデータから Word2Vec のモデルを作るところまでの方法を解説していきます。ただ、実際やってみてわかったのですが、Spark 自体、Spark MLlib の Word2Vec クラス、およびクラスタ構築に使った spark-ec2 に設定項目が多いせいで、細かいところで何度も何度もつまづきました……。

そのため、このシリーズでは各ステップについて、「最終的にやったこと」と、その最終的なやり方にたどり着くまでに「つまづいたこと」を分けました。やり方を知りたいだけの場合は「最終的にやったこと」の方だけ読んでください。「つまづいたこと」は、うまく行かなかった場合のための参考情報です。

Part 1 の範囲

Amazon EC2 に master 1台、slave 3台構成の Spark Cluster (Standalone mode) を構築し、spark-shell から Word2Vec を実行するところまで。

Spark をローカル環境（Mac）にインストールする

最終的にやったこと

まず、ローカル環境で Spark MLlib が動くかどうかを試してみました。環境は以下の通りです。

• MacBook Pro (Retina, 15-inch, Mid 2014)
• OS: OS X Yosemite 10.10.5
• CPU: 2.2 GHz Intel Core i7
• メモリ: 16GB 1600 MHz DDR3

OS X に Spark をインストールする場合、以下のコマンドだけでインストールできます（参考：ApacheSpark — BrewFormulas）。

% brew update
% brew install apache-spark

私が試した時点では Spark 1.6.1 でした。Java は Java 8 です。

% spark-shell --version
Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /___/ .__/\_,_/_/ /_/\_\   version 1.6.1
      /_/

Type --help for more information.

つまづいたこと

spark-shell ローカルモードで（--master local を指定して）実行すると、spark> というプロンプトが表示されるまでに、色々と WARN が出ます。

16/06/07 00:17:36 WARN Connection: BoneCP specified but not present in CLASSPATH (or one of dependencies)

BoneCP は JDBC Connection Pool ライブラリの名前です。scala - What do WARN messages mean when starting spark-shell? - Stack Overflow によると、ローカルモードで実行しているときは問題ないとのこと。

16/06/07 00:17:38 WARN ObjectStore: Version information not found in metastore. hive.metastore.schema.verification is not enabled so recording the schema version 1.2.0
16/06/07 00:17:38 WARN ObjectStore: Failed to get database default, returning NoSuchObjectException

こちらも、Hive metastore に接続できないことを表す WARN なので、ローカルモードでは関係ないと判断しました。

ローカル環境での Word2Vec の実行

最終的にやったこと

Spark MLlib のページ（Feature Extraction and Transformation）に、Spark MLlib に含まれる Word2Vec クラスを使ったサンプルコードがあります。これをローカルモードで実行してみます。

まず、サンプルコードで使っている text8.zip をダウンロードして、解凍します。これは、スペースで区切られた英単語が羅列された（意味のある文章ではない）100 MB のテキストファイルです。

% wget http://mattmahoney.net/dc/text8.zip
% unzip text8.zip
% ls -la text8
-rw-r--r--@ 1 myoshiz  staff  100000000  6  9  2006 text8

この text8 を置いたディレクトリで、以下のコマンドを実行します。spark.driver.memory はドライバのメモリ使用量を表すオプションで、デフォルトは 1g （1GB）です。

% spark-shell --master local \
--conf spark.driver.memory=5g \
--conf spark.serializer=org.apache.spark.serializer.KryoSerializer

spark-shell のプロンプトで、Word2Vec のサンプルコード を入力すれば、myModelPath ディレクトリ以下に、Word2Vec のモデルデータが生成されます。

なお、spark-shell の起動中は http://localhost:4040/ にアクセスすることで、ジョブの状態を確認できます。

つまづいたこと

最初は --conf spark.driver.memory=5g" を指定せずに spark-shell を起動していました。その状態でword2vec.fit(input)` を実行すると、OutOfMemoryError で spark-shell が落ちます。私の環境では、ファイルが 100MB だと落ちて、80MB まで減らすと落ちない、という状態でした。

scala> val model = word2vec.fit(input)
[Stage 0:>                                                          (0 + 1) / 3]
Exception in thread "refresh progress" java.lang.OutOfMemoryError: GC overhead limit exceeded
    at scala.StringContext.s(StringContext.scala:90)
（スタックトレース、および後続のエラーは省略）

エラーメッセージをもとに調べたところ、JVM の設定が悪いような情報をいくつか見かけました。

• Solved: spark-shell java.lang.OutOfMemoryError: PermGen sp... - Cloudera Community
  • SPARK_REPL_OPTS="-XX:MaxPermSize=256m" spark-shell --master local と指定すればよい、という情報

しかし、これを指定してもエラーメッセージは変わりませんでした。というか、私は Java 8 で実行していたので、そもそもこの設定には意味がありませんでした。

• Java8のHotSpotVMからPermanent領域が消えた理由とその影響 | ギークを目指して
• Java 8をインストールしたら、sbtでMaxMetaspaceSizeを設定する？ - CLOVER

この Java の仕様変更を踏まえて、以下のように spark-shell を実行したところ、落ちなくなりました。ただし、この方法だと、OutOfMemoryError が出ないだけで、いつまでも処理が終わらないという状態になってしまいました……。

% SPARK_REPL_OPTS="-XX:MaxMetaspaceSize=1024m" spark-shell --master local

結局、Configuration に載っているメモリ関係のパラメータを一通り確認して、前述の spark.driver.memory を増やしたところ、うまく動いたようで、処理が完了しました。JVM のパラメータを変更する必要はなかったようです。

Amazon EC2 への Spark クラスタの構築（spark-ec2 を使った方法）

最終的にやったこと

Slave の台数を増やすことで、Spark MLlib の実行時間が短くなることを確認するために、Spark クラスタを構築しました。今回は Spark に同梱されている spark-ec2 というスクリプトを使って構築しました。このスクリプトの説明は Running Spark on EC2 - Spark 1.6.1 Documentation にあります。

Amazon Elastic MapReduce (EMR) で Spark を使えることは知っていますが、いずれオンプレに Spark クラスタを構築したかったのと、かといってマシンスペックを何パターンか試すときに手作業での構築は大変すぎたので spark-ec2 を使いました。

まず、AWS のマネジメントコンソールを使って、以下の設定を行います。今回は Spark の話がメインなので、AWS の設定の詳細は省略します。

• IAM ユーザ "word2vec-user" の作成
• IAM ユーザ "word2vec-user" に対する "AdministratorAccess" ポリシーのアタッチ（EC2 と S3 に絞っても良い）
• EC2 でのキーペア "word2vec-key-pair" の作成
• ローカルマシンに対する環境変数 AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY の設定
• ローカルマシンに対するキーペアの配置（以下では /Users/myoshiz/.ssh/word2vec-key-pair.pem に置いたと仮定）

brew で Spark をインストールすると、spark-ec2 は入っていません。そのため、spark-ec2 を使うために、Apache Spark のダウンロードページ から zip ファイルをダウンロードします。今回は、以下のファイルを選択しました。

• Spark release: 1.6.1
• Package type: Pre-built for Hadoop 2.6 and later

ダウンロードした zip ファイルを解凍すると、ec2 ディレクトリに spark-ec2 というスクリプトが入っています。このディレクトリに移動し、まずは master 1台、slave 3台のクラスタを構築してみます。そのためには、以下のコマンドを実行します。

% ./spark-ec2 \
--key-pair=word2vec-key-pair \
--identity-file=/Users/myoshiz/.ssh/word2vec-key-pair.pem \
--region=us-west-1 \
--zone=us-west-1a \
--instance-type=m4.large \
--copy-aws-credentials \
--hadoop-major-version=yarn \
--slaves 3 \
launch spark-cluster

各オプションの意味と、上記の値を指定した理由は以下の通りです。

• --region は、デフォルトはバージニア北部（us-east-1）が使われる。国内だとインスタンス利用費が若干高く、東海岸は遠いので、北カリフォルニア（us-west-1）を指定した。
• --instance-type は、デフォルトでは m1.large が使われる。m1.large は古いインスタンスタイプのため、スペックに比して割高のため、同じく 2 vCPU、メモリ8GBの m4.large を指定。調べた時点では m1.large が $0.19/hour、m4.large が $0.14/hour だった。
• --copy-aws-credentials を指定すると、環境変数に設定された AWS のアクセスキーが、master の hadoop にも設定される。ただし、後述の通り Spark に対しては設定されないので、hadoop コマンドを使わないなら、指定しなくても良い。
• --hadoop-major-version=yarn は、使用する Hadoop のバージョンを指定する。今回は Pre-built for Hadoop 2.6 and later をダウンロードしているので、yarn を指定する必要がある。デフォルトは 1（Hadoop 1.0.4）。
• --slaves は slave の台数を指定する。

10〜20分待つとクラスタの構築が完了し、以下のようなメッセージが表示されます。Mac から以下の URL にアクセスすると、Spark UI や、Ganglia の画面を確認できます。

Spark standalone cluster started at http://ec2-xxx-xxx-xxx-xxx.us-west-1.compute.amazonaws.com:8080
Ganglia started at http://ec2-xxx-xxx-xxx-xxx.us-west-1.compute.amazonaws.com:5080/ganglia
Done!

上記のホスト名は、以降の作業でも使うので、以下のように環境変数に設定しておきます。シェルの設定ファイル（.bash_profile とか）で指定してもいいですが、クラスタを作るたびにホスト名が変わる点だけは注意が必要です。

% export EC2_SPARK_MASTER=ec2-xxx-xxx-xxx-xxx.us-west-1.compute.amazonaws.com

つまづいたこと(1)：GitHub の spark-ec2

brew で spark をインストールすると、そのなかには spark-ec2 が入っていません。そのため、このスクリプトだけ別に入手できないかと思い、GitHub で公開されている spark-ec2 を clone して実行してみました。

github.com

この spark-ec2 を実行すると、エラーも出ずに最後まで処理が進むのですが、Spark クラスタが起動しないようです。http://ec2-xxx-xxx-xxx-xxx.us-west-1.compute.amazonaws.com:8080 にアクセスしても応答がなく、spark-shell で --master spark://ec2-xxx-xxx-xxx-xxx.us-west-1.compute.amazonaws.com:7077 を指定しても接続できない、という状態になりました。

色々悩んだ結果、大人しく Apache Spark のダウンロードページ から zip ファイルをダウンロードして、そのなかの spark-ec2 を使ったところ、実行したコマンドの引数は同じにも関わらず、クラスタが起動しました。

GitHub 版も見た目はきちんと動いているように見えたために、他に原因があると思い込んでしまい、この問題で数日詰まってしまいました……。

つまづいたこと(2)：--hadoop-major-version=yarn の指定

このオプションは、公式サイトの Running Spark on EC2 には書かれていません。しかし spark-ec2 --help を実行すると、以下のオプションが表示されます。

  --hadoop-major-version=HADOOP_MAJOR_VERSION
                        Major version of Hadoop. Valid options are 1 (Hadoop
                        1.0.4), 2 (CDH 4.2.0), yarn (Hadoop 2.4.0) (default:
                        1)

上記のオプションを指定しないと、クラスタの構築後に spark-shell を実行した時に、以下のようなエラーが出て sqlContext の初期化に失敗しました。

16/06/13 14:08:02 INFO DataNucleus.Datastore: The class "org.apache.hadoop.hive.metastore.model.MResourceUri" is tagged as "embedded-only" so does not have its own datastore table.
java.lang.RuntimeException: java.io.IOException: Filesystem closed
    at org.apache.hadoop.hive.ql.session.SessionState.start(SessionState.java:522)
（中略）
<console>:16: error: not found: value sqlContext
         import sqlContext.implicits._
                ^
<console>:16: error: not found: value sqlContext
         import sqlContext.sql
                ^

Spark クラスタでの Word2Vec の実行

最終的にやったこと

spark-ec2 の login コマンドを使用すると、master にログインできます。もちろん ssh でもログインできますが、master のホスト名を書かなくてよいのがメリットだと思います。ちなみに、オプションの指定が面倒ですが、以下の3つは必須のようです。

% ./spark-ec2 \
--key-pair=word2vec-key-pair \
--identity-file=/Users/myoshiz/.ssh/word2vec-key-pair.pem \
--region=us-west-1 \
login spark-cluster

次に、先ほどと同じサンプルコードを実行するために、text8.zip をダウンロードします。また、このファイルを、クラスタ上で動作する HDFS にアップロードします。これは、ファイルを slave からアクセス可能にするための作業です。後ほど、HDFS の代わりに S3 を使う方法も紹介します。

$ wget http://mattmahoney.net/dc/text8.zip
$ unzip text8.zip
$ ./ephemeral-hdfs/bin/hadoop fs -put text8 /

ここまでの準備が終わったら、master 上で spark-shell を実行します。指定するオプションは以下の通りです。ローカルモードの場合とは、--master の指定が変わっています。

$ ./spark/bin/spark-shell \
--master spark://${EC2_SPARK_MASTER}:7077 \
--conf spark.driver.memory=5g \
--conf spark.serializer=org.apache.spark.serializer.KryoSerializer

ホスト名の指定が面倒ですが、--master spark://localhost:7077 という指定では接続できませんでした。

あとは、spark-shell で以下のように入力すると、Word2Vec が実行されます。ローカルモードとの違いは、textFile() や save() に渡されたファイルパスが、HDFS のファイルパスとして扱われることです。今回はルート直下に text8 を置いたため、/text8 のように指定しています。

import org.apache.spark._
import org.apache.spark.rdd._
import org.apache.spark.SparkContext._
import org.apache.spark.mllib.feature.{Word2Vec, Word2VecModel}

val input = sc.textFile("/text8").map(line => line.split(" ").toSeq)

val word2vec = new Word2Vec()

val model = word2vec.fit(input)

val synonyms = model.findSynonyms("china", 40)

for((synonym, cosineSimilarity) <- synonyms) {
  println(s"$synonym $cosineSimilarity")
}

// Save and load model
model.save(sc, "/model_text8")
val sameModel = Word2VecModel.load(sc, "/model_text8")

以上により、Word2Vec のジョブが slave 上で実行されます。ただ、http://ec2-xxx-xxx-xxx-xxx.us-west-1.compute.amazonaws.com:8080 にアクセスするとわかるのですが、このままだと3台ある slave のうち、1台しか使われません。次は、ジョブを分散するために、Word2Vec のパラメータを変更します。

つまづいたこと

最初、ローカルディスクのファイルにアクセスできないことに気づきませんでした。file:// を付けても駄目でした。

次に、HDFS 上にファイルをアップロードする方法で悩んだのですが、これは ./ephemeral-hdfs/bin 以下のコマンドが使えることに気付いたあとは簡単でした。hadoop コマンドに馴染みのない人は、Apache Hadoop 2.7.1 – などが参考になると思います。

すべての slave に処理が分散されることの確認（Word2Vec のパラメータ変更）

最終的にやったこと

Word2Vec のパラメータは、Word2Vec クラスの setter で指定できます。用意された setter とそのデフォルト値は Word2Vec の API リファレンス に記載されています。

これらの setter のうち、setNumPartitions() でパーティション数を1よりも大きくすると、複数の slave 間で処理が分散されます。この値のデフォルトが1なので、そのままでは slave が1台しか使われません。

val word2vec = new Word2Vec()

// Set this
word2vec.setNumPartitions(4)

val model = word2vec.fit(input)

slave 3台で試したところ、パーティション数を4まで増やした段階で、すべてのslaveに処理が分散されました。ただ、5台で試したときには、パーティション数を6にしても、slave 4台しか使われませんでした。単純に slave の台数 + 1 にすればよいというわけではなさそうで、詳細はまだわかりませんが、少なくとも slave の台数よりも大きい数を指定する必要がありそうです。

ただ、このパーティション数を増やすと、増やした分だけ負荷分散されて処理時間が短くなっていく一方で、計算結果の正確さも落ちていくとのことです。Word2Vec の処理が分散しない理由を調べている際に、以下の情報を見かけました。

stackoverflow.com

• イテレーションの数は、パーティション数と同じか、それ以下にすべき
• 正確さのために、パーティション数は小さい値を使うべき
• 結果（モデル）を正確にするためには、複数のイテレーションが必要

どれくらい結果が変わっていくのか、text8 を3台の slave 上で処理して調べてみました。以下は、numPartition = 1, 3, 6 での、"china" に類似した単語の上位10件です。パーティションが1個の場合の上位3件を太字にしています。text8 は意味のない文字列ですが、結果が変わっていく様子は参考になると思います。

numPartitions	1	3	6
1位	taiwan	taiwan	indonesia
2位	korea	korea	taiwan
3位	japan	japan	afghanistan
4位	mongolia	mainland	kazakhstan
5位	shanghai	indonesia	pakistan
6位	tibet	india	japan
7位	republic	pakistan	ireland
8位	india	mongolia	india
9位	manchuria	thailand	uzbekistan
10位	thailand	africa	iran

ちなみに、使われた slave の台数と、処理時間の関係は以下のようになりました。text8 くらいのデータ量（100 MB）だと、おおよそ、使われる slave の台数に応じて大きく処理時間が減るようです。

numPartitions	1	3	6
slave の台数	1	2	3
処理時間	6.3 min	3.6 min	1.9 min

つまづいたこと

負荷分散しない理由が Word2Vec のパラメータの方にある、ということが最初なかなか分からずに苦労しました。

普通に考えると「Spark MLlib から Word2Vec を使いたい人＝負荷分散を期待している人」だから、Word2Vec のパラメータの初期値は負荷分散するようになっているはずだ（だから Spark のパラメータの方に問題があるはずだ）と思い込んでいました……。

Part 1 のまとめ

ここまでの手順で、Amazon EC2 上に Spark クラスタを構築する方法を確認できました。

次の Part 2 では、この Spark クラスタの slave の台数およびマシンスペックを強化し、処理するデータ量も Wikipedia 英語版（Gzip 圧縮した状態で 12 GB）まで増やしてみます。

Part 1 の主な参考文献

• Word2Vec のサンプルコード
  • Feature Extraction and Transformation - spark.mllib - Spark 1.6.1 Documentation
• spark-ec2 の使い方
  • Running Spark on EC2 - Spark 1.6.1 Documentation
• API リファレンス
  • Word2Vec - org.apache.spark.mllib.feature.Word2Vec
  • Word2VecModel - org.apache.spark.mllib.feature.Word2VecModel