SparkアプリをElastic MapReduceでバッチ実行する

SparkをAWSのEMRで動かすチュートリアルとして下の2つを見つけた。

1. Run Spark and Shark on Amazon Elastic MapReduce - AWSの記事
2. Spark Example Project released for running Spark jobs on EMR - Snowplow社の記事

今回はSparkに絞ってバッチ実行のイメージを掴みたかったので、2つ目のチュートリアルを試した。

サンプル処理の内容はS3のバケットにあるテキストファイルの単語の出現頻度を集計してS3に保存するというもので、基本的には "2. How to use Spark Example Project" に従ってやればOKだった。下は、やってるうちに気づいたTips的なメモ。

安価なインスタンスの利用

このサンプル動かすだけなら記事にある m1.xlarge ほどパワーなくても良いので、elastic-mapreduceコマンドのオプションにある --instance-type m1.xlarge --instance-count 3の部分を下のように変えるとインスタンス代金を安いで気楽に試せた。

$ ./elastic-mapreduce ... \
 --instance-group master --instance-type m1.small --instance-count 1 --bid-price 0.03 \
 --instance-group core --instance-type m1.small --instance-count 2 --bid-price 0.03 \
 ...

指定内容は、マスターノードとコアノードにそれぞれ1, 2個のインスタンスを割当て、いずれも --bid-price をつけてスポットインスタンスを使う。入札最高価格はEC2コンソールの価格履歴から余裕を持って現在価格の3倍ほど($0.03/h)で設定した。

S3のアクセス制限

S3にアップロードしたjarを公開する(make the file publically accessible) という記載があったけれど、jarを公開しなくとも動作した。これはマスターノードの/home/hadoop/spark/conf/core-site.xml に、elastic-mapreduceが読むcredentials.jsonと同じAWSキーがセットされるためかと思う。

デバッグ

コマンドに --enable-debugging をつけると、S3に保存されるログに task-attempts というディレクトリが追加され、出力される情報が少し増える。

S3のログからもエラーの原因が分からない時は、ノードにログインして調べる。

このために、コマンドの--step-action TERMINATE_JOB_FLOWの部分を下に変えて、エラー時にインスタンスを起動したままにする。

$ ./elastic-mapreduce ... \
  --step-action CANCEL_AND_WAIT \
  --alive \
  ...

起動したら、コマンドの--ssh オプションでマスターノードにログイン。

$ ./elastic-mapreduce --ssh -j {job-id}

S3に保存されるジョブ実行ログの {job-id}/steps/1/controller ファイルにある INFO Executing /usr/java/latest/bin/java -cp /home/hadoop/conf:... のような行を見つけ、マスターノード上でその行のコマンドを実行して出力に異常がないか確認する。

※ クラスタのデバッグが終わったら --terminate で終了させる。

./elastic-mapreduce --terminate -j {job-id}

コマンドのテンプレ

ということでS3入出力のちょっとしたSparkアプリをEMRで動かすには、下のようにすると安価に試せるのでは：

$ ./elastic-mapreduce --create --name "Spark Example Project"  \
  --instance-group master --instance-type m1.small --instance-count 1 --bid-price 0.03 \
  --instance-group core --instance-type m1.small --instance-count 2 --bid-price 0.03 \
  --bootstrap-action s3://elasticmapreduce/samples/spark/0.8.1/install-spark-shark.sh --bootstrap-name "Install Spark/Shark" \
  --jar s3://elasticmapreduce/libs/script-runner/script-runner.jar --step-name "Run Spark Example Project" \
  --step-action TERMINATE_JOB_FLOW \
  --arg s3://snowplow-hosted-assets/common/spark/run-spark-job-0.1.0.sh \
  --arg s3://MY_BUCKET/spark/spark-example-project-0.2.0.jar \
  --arg com.snowplowanalytics.spark.WordCountJob \
  --arg s3n://MY_BUCKET/spark/hello.txt \
  --arg s3n://MY_BUCKET/spark/results

MY_BUCKETは適当なバケット名に、WordCountJob以外のアプリを作る場合は spark-example-project-0.2.0.jarとcom.snowplowanalytics.spark.WordCountJobの部分を変える。

その他 Reference

Spark on EMRについて、下の記事も気になった：

• http://www.slideshare.net/n3104/run-spark-on-emr Spark on EMRの仕組みとチューニングについて
• http://stackoverflow.com/questions/25420861/aws-emr-and-spark-1-0-0 Sparkの新しい版(1.0.0)を使いたいときの注意事項