本文是一个如何使用Spark的简要教程。首先通过Spark的交互式Shell来介绍API（使用Python或Scala），然后展示如何用Java，Scala和Python来写Spark引用程序。更完整的内容请参考编程指南。
为了跟随这篇指南进行学习，请首先从Spark网站上下载一个Spark release包。因为我们这里不会使用HDFS，所以下载一个用于任意Hadoop版本的包即可。
使用Spark Shell进行交互式分析
基础
Spark的Shell提供了一种学习API的简单方法，同样也是交互式分析数据的强大工具。在Scala（运行在Java虚拟机上，可以使用现有的Java库）或Python中可用。在Spark目录下运行下面命令：
./bin/spark-shell
Spark的主要抽象是弹性分布式数据集（RDD）。RDD可以通过Hadoop输入格式（比如HDFS文件）或者其它RDD转换来创建。让我们使用Spark源路径的README文件的文本内容来创建一个RDD。
scala> val textFile = sc.textFile("README.md")
textFile: org.apache.spark.rdd.RDD[String] = README.md MapPartitionsRDD[1] at textFile at <console>:25
RDD有可以返回值的actions，有可以返回指向新RDD指针的transformations。让我们从几个actions开始：
scala> textFile.count() // Number of items in this RDD
res0: Long = 126 // May be different from yours as README.md will change over time, similar to other outputs
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: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[2] at filter at <console>:27
可以把transformations和actions链接起来使用。
scala> textFile.filter(line => line.contains("Spark")).count() // How many lines contain "Spark"?
res3: Long = 15
更多RDD操作
RDD的actions和transformations可用于更复杂的计算。如我们想找到单词最多的行：
scala> textFile.map(line => line.split(" ").size).reduce((a, b) => if (a > b) a else b)
res4: Long = 15
首先使用map函数将每一行转换成一个整数，创建了一个新RDD。然后在这个新RDD上调用reduce函数找出最大行的单词数。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
一种常见的数据流模式为MapReduce，是由Hadoop推广的。Spark可以很容易地实现MapReduce：
scala> val wordCounts = textFile.flatMap(line => line.split(" ")).map(word => (word, 1)).reduceByKey((a, b) => a + b)
wordCounts: org.apache.spark.rdd.RDD[(String, Int)] = ShuffledRDD[8] at reduceByKey at <console>:28
这里我们结合了flatmap，map和reduceByKey转换来计算文件中每个单词的数量，生成一个(String，Int)对的RDD。
为了在我们的Shell中统计单词数量，可以使用collect action：
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> linesWithSpark.cache()
res7: linesWithSpark.type = MapPartitionsRDD[2] at filter at <console>:27
scala> linesWithSpark.count()
res8: Long = 15
scala> linesWithSpark.count()
res9: Long = 15
用Spark来探索和缓存一个100行的文本文件看起来有点呆。有意思的是这些相同的函数可以用在非常庞大的数据集上，即使跨越数十或者数百个节点。你也可以像编程指南中描述的一样通过使用bin/spark-shell连接到集群的方式，来交互式地做这件事。
独立的应用程序
假设我们想用Spark API写一个独立的应用程序。可以使用Scala（with sbt），Java（with Maven）和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(s"Lines with a: $numAs, Lines with b: $numBs")
sc.stop()
}
}
可以看到应用程序应该定义main()函数而不是扩展scala.App。使用scala.App的子类可能会运行不正常。这个程序就计算了Spark的README文件中包含a的行数和包含b的行数。注意把YOUR_SPARK_HOME替换成Spark的安装路径。不像之前使用Spark Shell的例子会初始化自己的SparkContext，我们需要初始化SparkContext作为程序的一部分。
我们要传一个包含了应用程序信息的SparkConf对象给SparkContext构造函数。
应用程序需要依赖Spark API，所以会包含一个sbt配置文件，simple.sbt，里面说明了Spark是一个依赖。这个文件也添加了一个Spark依赖的库：
name := "Simple Project"
version := "1.0"
scalaVersion := "2.11.7"
libraryDependencies += "org.apache.spark" %% "spark-core" % "2.1.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.11/simple-project_2.11-1.0.jar
# Use spark-submit to run your application
$ YOUR_SPARK_HOME/bin/spark-submit \
--class "SimpleApp" \
--master local[4] \
target/scala-2.11/simple-project_2.11-1.0.jar
...
Lines with a: 46, Lines with b: 23
从这里开始
恭喜你运行了第一个Spark应用程序！
- 查看API的深入概述，从Spark编程指南开始，或者看其它组件的编程指南菜单。
- 要在集群上运行应用程序，请看部署概述。
- 最后，Spark在examples目录中包含了很多示例（Scala，Java，Python，R），可以使用如下方式运行：
# 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
