Spark Shell快速入门

本教程是对Spark的一个快速简介。首先，我们通过Spark的交互式shell介绍一下API（主要是Python或Scala），然后展示一下如何用Java、Scala、Python写一个Spark应用。

首先，请到Spark website下载一个Spark发布版本，以便后续方便学习。我们暂时还不会用到HDFS，所以你可以使用任何版本的Hadoop。

1 使用Spark shell交互式分析

基础

利用Spark shell 很容易学习Spark API，同时也Spark shell也是强大的交互式数据分析工具。Spark shell既支持Scala（Scala版本的shell在Java虚拟机中运行，所以在这个shell中可以引用现有的Java库），也支持Python。在Spark目录下运行下面的命令可以启动一个Spark shell：

• Scala
• Python

./bin/spark-shell

Spark最主要的抽象概念是个分布式集合，也叫作弹性分布式数据集（Resilient Distributed Dataset – RDD）。RDD可以由Hadoop InputFormats读取HDFS文件创建得来，或者从其他RDD转换得到。下面我们就先利用Spark源代码目录下的README文件来新建一个RDD：

scala> val textFile = sc.textFile("README.md")
textFile: spark.RDD[String] = spark.MappedRDD@2ee9b6e3

RDD有两种算子，action算子（actions）返回结果，transformation算子（transformations）返回一个新RDD。我们先来看一下action算子：

scala> textFile.count() // Number of items in this RDD
res0: Long = 126

scala> textFile.first() // First item in this RDD
res1: String = # Apache Spark

再来看下如何使用transformation算子。我们利用filter这个transformation算子返回一个只包含原始文件子集的新RDD。

scala> val linesWithSpark = textFile.filter(line => line.contains("Spark"))
linesWithSpark: spark.RDD[String] = spark.FilteredRDD@7dd4af09

把这两个例子串起来，我们可以这样写：

scala> textFile.filter(line => line.contains("Spark")).count() // How many lines contain "Spark"?
res3: Long = 15

更多RDD算子

RDD action 和 transformation 算子可以做更加复杂的计算。下面的代码中，我们将找出文件中包含单词数最多的行有多少个单词：

• Scala
• Python

scala> textFile.map(line => line.split(" ").size).reduce((a, b) => if (a > b) a else b)
res4: Long = 15

首先，用一个map算子将每一行映射为一个整数，返回一个新RDD。然后，用reduce算子找出这个RDD中最大的单词数。map和reduce算组的参数都是scala 函数体（闭包），且函数体内可以使用任意的语言特性，或引用scala/java库。例如，我们可以调用其他函数。为了好理解，下面我们用Math.max作为例子：

scala> import java.lang.Math
import java.lang.Math

scala> textFile.map(line => line.split(" ").size).reduce((a, b) => Math.max(a, b))
res5: Int = 15

Hadoop上的MapReduce是大家耳熟能详的一种通用数据流模式。而Spark能够轻松地实现MapReduce流程：

scala> val wordCounts = textFile.flatMap(line => line.split(" ")).map(word => (word, 1)).reduceByKey((a, b) => a + b)
wordCounts: spark.RDD[(String, Int)] = spark.ShuffledAggregatedRDD@71f027b8

这个例子里，我使用了flatMap, map, and reduceByKey 这几个transformation算子，把每个单词及其在文件中出现的次数转成一个包含（String，int）键值对的RDD，计算出每个单词在文件中出现的次数：

scala> wordCounts.collect()
res6: Array[(String, Int)] = Array((means,1), (under,2), (this,3), (Because,1), (Python,2), (agree,1), (cluster.,1), ...)

缓存

Spark同样支持把数据集拉到集群范围的内存缓存中。这对于需要重复访问的数据非常有用，比如：查询一些小而”热“（频繁访问）的数据集 或者 运行一些迭代算法（如 PageRank）。作为一个简单的示例，我们把 linesWithSpark 这个数据集缓存一下：

• Scala
• Python

scala> linesWithSpark.cache()
res7: spark.RDD[String] = spark.FilteredRDD@17e51082

scala> linesWithSpark.count()
res8: Long = 19

scala> linesWithSpark.count()
res9: Long = 19

用Spark来缓存一个100行左右的文件，看起来确实有点傻。但有趣的是，同样的代码可以用于缓存非常大的数据集，即使这些数据集可能分布在数十或数百个节点，也是一样。你可以用 bin/spark-shell 连到一个集群上来验证一下。

独立的应用程序

假设我们想写一个独立的Spark应用程序。我们将快速的过一下一个简单的应用程序，分别用Scala（sbt编译），Java（maven编译）和Python。

• Scala
• Java
• Python

首先用Scala新建一个简单的Spark应用 – 简单到连名字都叫SimpleApp.scala

/* SimpleApp.scala */
import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
import org.apache.spark.SparkConf

object SimpleApp {
  def main(args: Array[String]) {
    val logFile = "YOUR_SPARK_HOME/README.md" // Should be some file on your system
    val conf = new SparkConf().setAppName("Simple Application")
    val sc = new SparkContext(conf)
    val logData = sc.textFile(logFile, 2).cache()
    val numAs = logData.filter(line => line.contains("a")).count()
    val numBs = logData.filter(line => line.contains("b")).count()
    println("Lines with a: %s, Lines with b: %s".format(numAs, numBs))
  }
}

注意，应用程序需要定义一个main方法，而不是继承scala.App。scala.App的子类可能不能正常工作。

这个程序，统计了Spark README文件中分别包含‘a’和’b’的行数。注意，你需要把YOUR_SPARK_HOME换成你的Spark安装目录。与之前用spark-shell不同，这个程序有一个单独的SparkContext对象，我们初始化了这个SparkContext对象并将其作为程序的一部分。

我们把一个 SparkConf 对象传给SparkContext的构造函数，SparkConf对象包含了我们这个应用程序的基本信息和配置。

我们的程序需要依赖Spark API，所以我们需要包含一个sbt配置文件，simple.sbt，在这个文件里，我们可以配置Spark依赖项。这个文件同时也添加了Spark本身的依赖库：

name := "Simple Project"

version := "1.0"

scalaVersion := "2.10.5"

libraryDependencies += "org.apache.spark" %% "spark-core" % "1.6.0"

为了让sbt能正常工作，我们需要一个典型的目录结构来放SimpleApp.scala和simple.sbt程序。一旦建立好目录，我们就可以创建一个jar包，然后用spark-submit脚本运行我们的代码。

# Your directory layout should look like this
$ find .
.
./simple.sbt
./src
./src/main
./src/main/scala
./src/main/scala/SimpleApp.scala

# Package a jar containing your application
$ sbt package
...
[info] Packaging {..}/{..}/target/scala-2.10/simple-project_2.10-1.0.jar

# Use spark-submit to run your application
$ YOUR_SPARK_HOME/bin/spark-submit \
  --class "SimpleApp" \
  --master local[4] \
  target/scala-2.10/simple-project_2.10-1.0.jar
...
Lines with a: 46, Lines with b: 23

2 下一步

恭喜你！你的首个Spark应用已经跑起来了！最后，Spark examples子目录下包含了多个示例，你可以这样来运行这些例子：

# For Scala and Java, use run-example:
./bin/run-example SparkPi

# For Python examples, use spark-submit directly:
./bin/spark-submit examples/src/main/python/pi.py

# For R examples, use spark-submit directly:
./bin/spark-submit examples/src/main/r/dataframe.R