結構はまってしまったのでメモ
VPC内に作成したEMRでSparkを動かしました。サンプルはいろいろなところにそこそこあるのですが、どうもきちっと動くものがなく結構苦労してしまいました。
EMR
まずはVPC内にEMRを作成します。EMRのコンソール画面を開き、CreateClusterを押します。
VPC内に作成するためには上の方にあるGo to Advanced optionから進む必要があります。
VPCとサブネットを設定し、パーミッションなどを設定しクラスターを作成します。
計算プログラム
こんな感じでディレクトリ作成します
なおscala 2.10,jdk1.8を使用しています
s3にあらかじめbucketという名のbucketとその下にoutputディレクトリを作成し、EMRから触れるようにパーミッションを設定しておきます
1 2 3 4 5 6 7 8 9 10 11 12 13 14 | ├── build.sbt ├── project │ └── assembly.sbt ├── src │ ├── main │ │ ├── java │ │ ├── resources │ │ └── scala │ │ └── sample │ │ └── SparkPi.scala │ └── test │ ├── resources │ └── scala └── target |
build.sbt
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 | name := "emrscala" version := "0.0.1" scalaVersion := "2.10.5" libraryDependencies ++= Seq( ("org.apache.spark" %% "spark-sql" % "1.3.1"). exclude("org.mortbay.jetty", "servlet-api"). exclude("com.google.guava","guava"). exclude("org.apache.hadoop","hadoop-yarn-api"). exclude("commons-beanutils", "commons-beanutils-core"). exclude("commons-beanutils", "commons-beanutils"). exclude("commons-collections", "commons-collections"). exclude("commons-logging", "commons-logging"). exclude("org.spark-project.spark", "unused"). exclude("com.twitter", "parquet-encoding"). exclude("com.twitter", "parquet-column"). exclude("com.twitter", "parquet-hadoop-bundle"). exclude("org.datanucleus", "datanucleus-api-jdo"). exclude("org.datanucleus", "datanucleus-core"). exclude("org.datanucleus", "datanucleus-rdbms"). exclude("com.esotericsoftware.minlog", "minlog"), ("org.apache.spark" %% "spark-mllib" % "1.3.1"). exclude("org.mortbay.jetty", "servlet-api"). exclude("com.google.guava","guava"). exclude("org.apache.hadoop","hadoop-yarn-api"). exclude("commons-beanutils", "commons-beanutils-core"). exclude("commons-beanutils", "commons-beanutils"). exclude("commons-collections", "commons-collections"). exclude("commons-logging", "commons-logging"). exclude("org.spark-project.spark", "unused"). exclude("com.twitter", "parquet-encoding"). exclude("com.twitter", "parquet-column"). exclude("com.twitter", "parquet-hadoop-bundle"). exclude("org.datanucleus", "datanucleus-api-jdo"). exclude("org.datanucleus", "datanucleus-core"). exclude("org.datanucleus", "datanucleus-rdbms"). exclude("com.esotericsoftware.minlog", "minlog"), ("org.apache.spark" %% "spark-hive" % "1.3.1"). exclude("org.mortbay.jetty", "servlet-api"). exclude("com.google.guava","guava"). exclude("org.apache.hadoop","hadoop-yarn-api"). exclude("commons-beanutils", "commons-beanutils-core"). exclude("commons-beanutils", "commons-beanutils"). exclude("commons-collections", "commons-collections"). exclude("commons-logging", "commons-logging"). exclude("org.spark-project.spark", "unused"). exclude("com.twitter", "parquet-encoding"). exclude("com.twitter", "parquet-column"). exclude("com.twitter", "parquet-hadoop-bundle"). exclude("org.datanucleus", "datanucleus-api-jdo"). exclude("org.datanucleus", "datanucleus-core"). exclude("org.datanucleus", "datanucleus-rdbms"). exclude("com.esotericsoftware.minlog", "minlog"), ("org.apache.spark" %% "spark-sql" % "1.3.1"). exclude("org.mortbay.jetty", "servlet-api"). exclude("com.google.guava","guava"). exclude("org.apache.hadoop","hadoop-yarn-api"). exclude("commons-beanutils", "commons-beanutils-core"). exclude("commons-beanutils", "commons-beanutils"). exclude("commons-collections", "commons-collections"). exclude("commons-logging", "commons-logging"). exclude("org.spark-project.spark", "unused"). exclude("com.twitter", "parquet-encoding"). exclude("com.twitter", "parquet-column"). exclude("com.twitter", "parquet-hadoop-bundle"). exclude("org.datanucleus", "datanucleus-api-jdo"). exclude("org.datanucleus", "datanucleus-core"). exclude("org.datanucleus", "datanucleus-rdbms"). exclude("com.esotericsoftware.minlog", "minlog"), ("org.apache.spark" %% "spark-core" % "1.3.1"). exclude("org.mortbay.jetty", "servlet-api"). exclude("com.google.guava","guava"). exclude("org.apache.hadoop","hadoop-yarn-api"). exclude("commons-beanutils", "commons-beanutils-core"). exclude("commons-beanutils", "commons-beanutils"). exclude("commons-collections", "commons-collections"). exclude("commons-logging", "commons-logging"). exclude("org.spark-project.spark", "unused"). exclude("com.twitter", "parquet-encoding"). exclude("com.twitter", "parquet-column"). exclude("com.twitter", "parquet-hadoop-bundle"). exclude("org.datanucleus", "datanucleus-api-jdo"). exclude("org.datanucleus", "datanucleus-core"). exclude("org.datanucleus", "datanucleus-rdbms"). exclude("com.esotericsoftware.minlog", "minlog") ) |
assembly.sbt
1 2 3 | addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "0.14.1") addSbtPlugin("com.typesafe.sbteclipse" % "sbteclipse-plugin" % "4.0.0") |
SparkPi.scala
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 | package sample import scala.math.random import org.apache.spark.mllib.util.MLUtils import org.apache.spark._ object SparkPi { def main(args: Array[String]) { val conf = new SparkConf().setAppName("Spark Pi").setMaster("local[2]") val spark = new SparkContext(conf) val slices=2 val n = math.min(100000L * slices, Int.MaxValue).toInt // avoid overflow val count = spark.parallelize(1 until n, slices).map { i => val x = random * 2 - 1 val y = random * 2 - 1 if (x*x + y*y < 1) 1 else 0 }.reduce(_ + _) println("Pi is roughly " + 4.0 * count / n) val outputLocation = "s3n://bucket/output" val pi=4.0 * count / n val data=spark.makeRDD(Seq(pi)) data.saveAsTextFile(outputLocation + "/pi") spark.stop() } } |
make
1 | sbt assembly |
できたjarファイルをS3にコピーします
実行
EMRのadd StepからCustom JARを選択
JARLocationに先ほどアップしたJarを選択
Argumentに
1 | --verbose sample.SparkPi |
こんな感じで実行
しばらくたつと s3://bucket/output/piいかに結果が格納されています。