はじめに

EMR上にSpark環境を構築して触っていたのですが、常時立ち上げて置くのもコストがかかるし
使わなくなったら消して、使うときになったら環境構築をしてというのが大変だったので
EC2上にSpark環境を構築しました。
好きなときにサーバの起動・停止が出来るので低コストでSparkの検証が行えます。
また分析が行いやすいようiPython Notebookを入れ、そこでSparkも扱えるようにしました。

EC2インスタンスの準備

Sparkに使うサーバの起動、停止、削除などを行なうくらいなので低スペックのもので十分です。

今回は一番安い、t2.microを使用しました。

Sparkのダウンロード

gitコマンドを使用するので
まずgitをインストールし必要なファイルを持ってきます。

sudo yum install -y git
git clone git://github.com/apache/spark.git -b branch-1.2

現時点(2015年1月)で最新だった1.2を入れています。

【参考】
Spark Lightning-fast cluster computing

AWSの設定

spark/ec2/spark_ec2.py を見ると
.botoでAWSの設定がされていればそこから、されていなければ環境変数のAWS_ACCESS_KEY_IDとAWS_SECRET_ACCESS_KEYなどを参照するようです。
今回はbotoの設定ファイルにアクセス・シークレットキーを設定しておきました。

~/.boto

[Credentials]
aws_access_key_id = <AWS_ACCESS_KEY_ID>
aws_secret_access_key = <AWS_SECRET_ACCESS_KEY>

Sparkクラスタの立ち上げ

~/spark/ec2/spark-ec2 --key-pair=<KEY_PAIR_NAME> --identity-file=<SSH_KEY_FILE_NAME> --region=ap-northeast-1 --spark-version=1.2.0 --copy-aws-credentials launch <CLUSTER_NAME>

【備考】
立ち上げに数分かかります。今回は5分ほどかかりました。
インスタンスはデフォルトだとm1.largeを2台が使われます。
「--copy-aws-credentials」これをつけるとAWSの設定が引き継がれます。

細かい設定に関しては下記を参照してください。
Running Spark on EC2
http://spark.apache.org/docs/1.2.0/ec2-scripts.html

セキュリティグループについて

Sparkクラスタを立ち上げると、自動でセキュリティグループも作成されますが、デフォルトの状態だとポートによっては全開放されているので、必要に応じて変更する必要があります。

また、iPython Notebookで8888番と9000番ポートを使用するのでアクセス出来るように追加をしてください。

サーバへログイン

~/spark/ec2/spark-ec2 --key-pair=<KEY_PAIR_NAME> --identity-file=<SSH_KEY_FILE_NAME> --region=ap-northeast-1 login <CLUSTER_NAME>

クラスタの停止は

~/spark/ec2/spark-ec2 --region=ap-northeast-1 stop <CLUSTER_NAME>

クラスタの起動は

~/spark/ec2/spark-ec2 --key-pair=<KEY_PAIR_NAME> --identity-file=<SSH_KEY_FILE_NAME> --region=ap-northeast-1 start <CLUSTER_NAME>

pysparkの実行

マスターサーバにログイン後以下のコマンドで実行出来ます。

/root/spark/bin/pyspark

scalaは

/root/spark/bin/spark-shell

iPython Notebookの設定

profileの作成

ipython profile create myserver

設定ファイルの編集

~/.ipython/profile_myserver/ipython_notebook_config.py

c = get_config()
c.NotebookApp.ip = '*' # もしくはマスターのローカルIP
c.NotebooksApp.open_browser = False

~/.ipython/profile_myserver/startup/00-myserver-setup.py

import os
import sys

os.environ['SPARK_HOME'] = '/root/spark/'
CLUSTER_URL = open('/root/spark-ec2/cluster-url').read().strip()
spark_home = os.environ.get('SPARK_HOME', None)
if not spark_home:
    raise ValueError('SPARK_HOME environment variable is not set')
sys.path.insert(0, os.path.join(spark_home, 'python'))
sys.path.insert(0, os.path.join(spark_home, 'python/lib/py4j-0.8.2.1-src.zip'))
execfile(os.path.join(spark_home, 'python/pyspark/shell.py'))
execfile(os.path.join(spark_home, 'conf/spark-env.sh'))

【参考】
How-to: Use IPython Notebook with Apache Spark

iPython Notebookの起動

ipython notebook --pylab inline --profile=myserver

< Sparkマスターサーバのドメイン:8888>にアクセスすると使えます。

ファイルの読み込み

【S3から読み込み、RDDとして扱う】

s3_file = sc.textFile("s3n://<BUCKET>/<DIR>")
s3_file = sc.textFile("s3n://<BUCKET>/<DIR>/<FILE>")

注意すべき点は「s3n://」と書くこと。「s3://」では読み込みが出来ませんでした。
ファイル単体はもちろんのこと、フォルダ参照もすることが出来ます。

【ローカルファイルを読み込む】

local_file = sc.textFile("file://" + "ローカルファイルパス")

ファイル名を指定する際、デフォルトではHDFSを参照するようです。

ファイルをHDFSで管理すれば良いのですが、
ローカルファイルを参照したいという時はパスの前に"file://"をつける必要があります。

更にiPython Notebookからローカルファイル参照をする場合
9000ポートを使用するので、セキュリティグループを必要に応じて開ける必要があります。

【S3から読み込み、DataFrameとして扱う】

これに関してはSparkは特に関係ないですが一応書いておきます。
pandasでS3のファイルを扱う場合はbotoの設定が必要です。