Spark on EMR(YARN対応)を動かす

はじめに

ほとんどのデータをS3に突っ込んでいる場合、アドホックな分析であったり統計モデル作成をしたりする際に、適当にバケットまるっとロードしてSparkSQLでクエリなげたりMlibに突っ込むというのは割りとよくあることかと思います。

EMRでSparkを動かす際に、これまではYARNに対応しているBootstrap Actionが無かったので設定がだるかったのですが、awslabsがYARN対応のbootstrap actionsを出しているのでそれを試してみます。

• http://blogs.aws.amazon.com/bigdata/post/TxO6EHTHQALSIB/Getting-Started-with-Amazon-EMR-Bootstrap-Actions
• https://github.com/awslabs/emr-bootstrap-actions/tree/master/spark

事前にやること

動かすサーバに、iamロールが付与されていること前提。必要なパーミッションは、EMR、EC2、S3、IAMのフルアクセス。

aws cliの設定

aws cliのconfigで、emrを有効化

~/.aws/config

[preview]
emr=true

ついでにリージョンの設定をしてくおくと、--region ap-northeast-1とかいらなくなる。

EMRのデフォルトロールを作成

aws emr create-default-roles --region ap-northeast-1

いらないパーミッションがいっぱい(DynamoDB full accessとか)付いてくるので適宜消す。

使ってみる

起動コマンド

東京リージョンの適当なホストのec2-userで試す。spotインスタンスで安く使ってみる。とりあえず1時間0.03ドルで試す。EMR料金と合わせて、一台一時間0.07ドル。とりあえずMaster 1台、Worker 10台で起動する。

aws emr create-cluster --region ap-northeast-1 --name SparkCluster --ami-version 3.3.1 --no-auto-terminate --service-role EMR_DefaultRole --instance-groups InstanceCount=1,BidPrice=0.03,Name=sparkmaster,InstanceGroupType=MASTER,InstanceType=m1.large InstanceCount=10,BidPrice=0.03,Name=sparkworker,InstanceGroupType=CORE,InstanceType=m1.large --ec2-attributes InstanceProfile=EMR_EC2_DefaultRole,KeyName={{KeyPairName}} --applications Name=HIVE --bootstrap-actions Path=s3://support.elasticmapreduce/spark/install-spark,Args=[-v,1.2]

起動すると、cluster idが返却される。これを忘れると大変。一応、list-clusterで見えるけど。

{{KeyPairName}}は、AWSに登録しているキーペアの名前を入れる。新しく作っても良い。作る場合はこんな感じ。秘密鍵は*.pemという名前で保存

aws ec2 create-key-pair --region ap-northeast-1 --key-name emr | jq -r '.KeyMaterial' | tee ~/.ssh/emr.pem && chmod 600 ~/.ssh/emr.pem

クラスタの確認とか

aws emr list-clusters --region ap-northeast-1

クラスタへのsshログイン

aws emr ssh --cluster-id クラスタID --key-pair-file ~/.ssh/emr.pem --region ap-northeast-1

key-pair-fileは拡張子が.pemとかじゃないと怒られる。

クラスタの停止

aws emr terminate-clusters --cluster-ids クラスタID --region ap-northeast-1

spark shellを起動する

10台2coreなので、以下のように起動。ワーカー毎にメモリ確保するので2gずつで切る。

./spark/bin/spark-shell --master yarn-client --num-executors 10 --driver-memory 2g --executor-memory 2g --executor-cores 2

とりあえず、fluent-plugin-s3で上げた1日分のnginxログをSparkSQLのtableにしてクエリ投げてみる。一行一jsonならJsonFileを使うだけで良いのだけれど、fluent-plugin-s3の出力はそうではないのでjson部分だけ取り出してJsonRDDでtable化する。

scala> var data = sc.textFile("s3://buket_name/hogehoge/20141122/*/*.gz")
scala> var jsons = data.map(line => line.split("\t")(2))
scala> val sqlContext = new org.apache.spark.sql.SQLContext(sc)
scala> var table = sqlContext.jsonRDD(jsons)
scala> table.printSchema()
root
 |-- agent: string (nullable = true)
 |-- code: string (nullable = true)
 |-- forwarded_for: string (nullable = true)
 |-- host: string (nullable = true)
 |-- method: string (nullable = true)
 |-- path: string (nullable = true)
 |-- referer: string (nullable = true)
 |-- response_time: string (nullable = true)
 |-- size: string (nullable = true)
 |-- user: string (nullable = true)

テーブルとして登録

scala> table.registerTempTable("logs")
scala> sqlContext.cacheTable("logs")

cacheTableで、メモリに載せたままにしてくれる様子。

クエリ投げてみる。数字は割愛

scala> var code_counts = sqlContext.sql("SELECT code,count(*) FROM logs GROUP BY code").collect()
scala> code_counts.foreach(println)

[400,26]
[404,...]
[405,...]
[416,...]
[304,...]
[200,...]
[206,...]
[499,...]

うむ。

hiveContext使うほうが高機能。sqlContextだとregexp_extract等のhiveのコマンドが使えない。とりあえず、paramsというgetパラメータの値毎にstatus codeを集計してみるとこんな感じ。

scala> val hiveContext = new org.apache.spark.sql.hive.HiveContext(sc)
scala> var htable = hiveContext.jsonRDD(jsons)
scala> htable.registerTempTable("logs")
scala> hiveContext.cacheTable("logs")
scala> var media_code = hiveContext.sql("SELECT a.param, a.code, count(*) FROM (SELECT regexp_extract(path, 'params=([0-9]+)', 1) AS param, code FROM logs WHERE path LIKE '%_show%') a GROUP BY a.param, a.code ORDER BY a.media, a.code")
scala> media_code.collect()
res3: Array[org.apache.spark.sql.Row] = Array([,200,...], ...

うむ。index効かせてゴリゴリjoinしないただのフルスキャンならload時間含めて4台クラスタ程度でもredshiftよりだいぶ早い感じ。一日分しかメモリに載せないし。

subqueryにくくり出さないとparamをGROUP BYに使えなかった。

バッチにする

--jar s3://elasticmapreduce/libs/script-runner/script-runner.jarをわたして、実行するバッチの内容が記載されたshell scriptを渡せば一番簡単っぽい。こちらを以下を参考にする。

• http://snowplowanalytics.com/blog/2014/04/17/spark-example-project-released/

わりかしダルいしscalaマンでは無いのでspark-shellにスクリプト食わしてやっていきたい。

pysparkを使ってみる

YARNの上でpysparkを動かすには環境変数"SPARK_YARN_USER_ENV"にpyspark用のコードを含んだPYTHONPATHを指定して置く必要がある。SPARK_YARN_USER_ENVは、YARNで動かしているslave系にも伝播して渡すことが出来る様子。

以下のコードで環境変数を設定可能

import os, sys, glob
SPARK_HOME = "/home/hadoop/spark/"
sys.path.insert(0, os.path.join(SPARK_HOME, 'python'))
for lib in glob.glob(os.path.join(SPARK_HOME, 'python/lib/py4j-*-src.zip')):
    sys.path.insert(0, lib)

SPARK_YARN_USER_ENV =  "PYTHONPATH=" + ":".join(sys.path)
if os.getenv('SPARK_YARN_USER_ENV', None):
    os.environ['SPARK_YARN_USER_ENV'] = os.environ['SPARK_YARN_USER_ENV'] + ",PYTHONPATH=" + ":".join(sys.path)
else:
    os.environ['SPARK_YARN_USER_ENV'] = "PYTHONPATH=" + ":".join(sys.path)

sshでクラスタにログインしたら、以下を実行してpysparkのシェルを起動。ipythonで使えないと不便なのでpipでipython入れる。

sudo pip install ipython[all]
MASTER=yarn-client IPYTHON=1 spark/bin/pyspark --master yarn-client --num-executors 10 --driver-memory 2g --executor-memory 2g --executor-cores 2

並列度の指定はspark-shellの起動と同じように指定すれば良い。

ipython notebookを使うにはまずprofileを設定する

ipython profile create spark

プロファイルには以下のように設定

~/.ipython/profile_spark/ipython_notebook_config.py

c = get_config()

c.IPKernelApp.pylab = 'inline'
c.NotebookApp.ip = '*'
c.NotebookApp.open_browser = False
c.NotebookApp.port = 8888

パスワードもきちんと設定したほうが良さそう。その上で、SecurityGroupのElasticMapReduce-masterの8888番を適切に開けると外部から接続できるようになる。

ipython notebookの起動は以下のコマンドで可能。作成したクラスタのmaster serverのglobal ipを持ってきて、8888番を開けばそのままpysparkが利用できる。

MASTER=yarn-client IPYTHON_OPTS='notebook --profile=spark' spark/bin/pyspark --master yarn-client --num-executors 10 --driver-memory 2g --executor-memory 2g --executor-cores 2

GraphXのPython APIが無いとはいえ、mllibも全部使えるようになっているし、1.2からmllib.featureが使えるようになって更に嬉しい。

このあたりの起動設定やらipythonのインストールやらはbootstrap actionsにまとめておくと良い。