Sparkのローカル開発環境作る方法（Scala向け） #Scala

Scalaのパッケージ管理としてsbtがあるらしいが、そもそもScala初めてで何もわからなかったので、自分の備忘録として。
せっかくだしEclipseからIntelliJに移行しようと思ったので、IntelliJベースで話進めます。Eclipserは適宜読み替えてもらえると。

sbt

Scalaのパッケージ管理。Rubyのgemみたいにグローバル汚さないで済むし（Bundler使えば良い話だが）便利。
機能的には幾つかあるらしいが、自分は依存関係の整理にしかまだ使ってない（使いこなせてない）のでいいプラグインあったら教えて下さい。
Sparkの各種ライブラリもここで管理する

IntelliJでプロジェクトを作る

New -> Projectで新規プロジェクトを作成する
Scala -> sbtを選択する。もしこの選択肢がなかったら環境設定からScalaのプラグインをインストールし再起動すると出てくる。
適当に名前入れてfinish.勝手にプロジェクトができる。
生成されるbuild.sbtに以下の項目を追記する
バージョンは適宜指定すること。

build.sbt

libraryDependencies ++= Seq(
  "org.apache.spark" %% "spark-core" % "1.4.1",
  "org.apache.spark" %% "spark-graphx" % "1.4.1",
  "org.apache.spark" %% "spark-sql" % "1.4.1"
)

保存してsbt実行。auto importにチェック入れておけば保存した瞬間に依存関係の整理が走る
正常終了すると、ディレクトリが何個かできる。この中のsrc/scala以下にパッケージ切ってコード書けばOK。試しにGraphX使ってみる。PageRankを求めてみる。

example.scala

package hellospark

import org.apache.spark.graphx.{Graph, GraphLoader}
import org.apache.spark.{SparkConf, SparkContext}

/**
 * Created by nishimuuu on 2015/08/16.
 */
object example {
  def main(args: Array[String]) {
    val conf = new SparkConf().setAppName("appname").setMaster("local[*]")
    val sc   = new SparkContext(conf)

    val edge_path = "/path/to/edge.csv"
    val node_path = "/path/to/node.csv"
    val graph: Graph[Int, Int] = GraphLoader.edgeListFile(sc, edge_path).cache()

    val ranks = graph.pageRank(0.001).vertices
    val nodes = sc.textFile(node_path).map{line =>
      val fields = line.split(",")
      (fields(0).toLong, fields(1))
    }

    val ranksByNode = nodes.join(ranks).map {
      case(id, (node, rank)) => (node, rank)
    }
    println(ranksByNode.collect().mkString("\n"))


    sc.stop()
  }
}

結果

(follower2 follow1 A,0.3300364712764322)
(follower2 follow0,0.4854936516596302)
(follower3 follow3,0.6927675771080204)
(follower4 follow4,0.8482247574912182)
(follower2 follow2,0.5260649857706428)
(follower2 follow1 B,0.3300364712764322)

こんな感じ。ファイルの場合、edgeを読み込んでnodeを張り付けていく、のが楽。
まだ細かく触ってないけど、読み込み時にゴニョゴニョやる必要がないのを知れてよかった。