Apache Sparkを利用したWord Count

はじめに

以下の文章についてApache Sparkを利用してWord Countを実行するコードを書いていきます。

Apache Spark is a fast and general-purpose cluster computing system. It provides high-level APIs in Java, Scala, Python and R, and an optimized engine that supports general execution graphs. It also supports a rich set of higher-level tools including Spark SQL for SQL and structured data processing, MLlib for machine learning, GraphX for graph processing, and Spark Streaming.

Java、Scala、sbtのインストール

JavaとScalaとsbtをインストールします。
今回は以下のバージョンをインストールします。

$ java -version
java version "1.8.0_31"
Java(TM) SE Runtime Environment (build 1.8.0_31-b13)
Java HotSpot(TM) 64-Bit Server VM (build 25.31-b07, mixed mode)
$ scala -version
Scala code runner version 2.11.6 -- Copyright 2002-2013, LAMP/EPFL
$ sbt about
[info] Set current project to App (in build file:/Users/)
[info] This is sbt 0.13.8
[info] The current project is {file:/Users/} 1.0
[info] The current project is built against Scala 2.11.6
[info] Available Plugins: sbt.plugins.IvyPlugin, sbt.plugins.JvmPlugin, sbt.plugins.CorePlugin, sbt.plugins.JUnitXmlReportPlugin
[info] sbt, sbt plugins, and build definitions are using Scala 2.10.4

実装1

以下の内容をbuild.sbtとして保存します。
今回はApache Spark 1.4.1を利用します。

build.sbt

name := "App"
version := "1.0"
scalaVersion := "2.11.6"

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

次に以下の内容をmain.scalaとして保存します。

main.scala

import org.apache.spark.SparkContext

object Main {
  def main(args: Array[String]) {
    val sc = new SparkContext("local[*]", "App") // ローカルモード
    val rdd = sc.textFile("sample.txt")
      .flatMap(r => r.split(" "))
      .map(r => (r, 1))
      .cache
      .reduceByKey(_ + _)
      .collect.foreach { r => 
        println(r._1.mkString("", "", "") + "\t" + r._2)
      }

    sc.stop()
  }
}

flatMapは、要素を別の型に変換します。
次のmap(r => (r, 1)) で(Apache, 1) (Spark, 1) というペアにします。
reduceByKey(_ + _) でカウントアップします。
reduceByKey(_ + _) は　reduceByKey((x, y) => x + y) と等価です。

以下の内容をsample.txtとして保存します。

sample.txt

Apache Spark is a fast and general-purpose cluster computing system. It provides high-level APIs in Java, Scala, Python and R, and an optimized engine that supports general execution graphs. It also supports a rich set of higher-level tools including Spark SQL for SQL and structured data processing, MLlib for machine learning, GraphX for graph processing, and Spark Streaming.

実行1

実行方法は以下のとおりです。

$ sbt run

大量の実行ログの中にprintlnの内容が表示されます。

GraphX	1
Python	1
provides	1
is	1
R,	1
higher-level	1
general	1
fast	1
Java,	1
SQL	2
Apache	1
data	1
learning,	1
cluster	1
graph	1
execution	1
MLlib	1
Scala,	1
computing	1
supports	2
engine	1
set	1
rich	1
Streaming.	1
Spark	3
graphs.	1
general-purpose	1
APIs	1
that	1
a	2
high-level	1
including	1
optimized	1
in	1
system.	1
of	1
tools	1
also	1
structured	1
It	2
for	3
an	1
machine	1
and	5
processing,	2

実装2

次に複数行ある場合を考えます。
以下の内容をsample.txtとして保存します。

sample.txt

A
B
C
A
B
A

次に以下の内容をmain.scalaとして保存します。
実装1との違いはflatMapがなくなっただけです。

main.scala

import org.apache.spark.SparkContext

object Main {
  def main(args: Array[String]) {
    val sc = new SparkContext("local[*]", "App") // ローカルモード
    val rdd = sc.textFile("sample.txt")
      .map(r => (r, 1))
      .cache
      .reduceByKey(_ + _)
      .collect.foreach { r => 
        println(r._1.mkString("", "", "") + "\t" + r._2)
      }

    sc.stop()
  }
}

実行2

実行結果は以下のとおりです。

B	2
A	3
C	1