本项目是 Apache Spark1.6.1方文档的中文翻译版，之前翻译过spark1.5，由于网站有组织翻译Spark1.6.1所以我又重新翻译了一下，我翻译的这一章和spark1.5一样。本次翻译主要针对对Spark感兴趣和致力于从事大数据方法开发的人员提供有价值的中文资料，希望能够对大家的工作和学习有所帮助。

这个翻译本应该在二月份完成，由于一直出差加上加班导致短短一篇文章迟迟没有完成，在此想大家表示深深的歉意。同时也希望有能力的童鞋能够一起努力推动大数据在中国的发展，让天下没有难学的技术谢谢大家。

Quick Start

• 交互分析(Interactive Spark ShellAnalysis with the Spark Shell)
  • 基础(Basics)
  • RDD上更多操作(More on RDD Operations)
  • 缓存Caching
• 独立应用(Self-Contained Applications)
• 下一章(Where to Go from Here)

本篇文章介绍了Spark快速入门. 我们将第一次通过Spark的Shell介绍API(通过 Python 或Scala), 然后展示怎么通过Java,Scala和Python写应用. 看 编程指南了解更多信息.

跟着下面指导, 第一步下载安装包从 Spark官方网址. 我们不会使用HDFS,你可以下载一个匹配任何Hadoop版本的发行版本.

Spark Shell交互分析

基础

Spark Shell提供了一个简单学习API的方式, 同时它也是一个有用的交互分析工具. 它可以使用Scala(它是一种运行在Java虚拟机上并且可以使用Java依赖包)或者Python. 切换到Spark目录并启动按照以下方式启动它:

• Scala
• Python

./bin/spark-shell

Spark的主要抽象是一个分布式数据集被称为弹性分布式数据集 (RDD). RDDs可以通过Hadoop上面的输入格式( 例如 HDFS 文件)或者通过其他的RDD转换. 让我们通过Spark目录中的README文件创建一个新的RDD:

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

RDDs 有actions(我翻译成操作因子), 它能够返回值, 和transformations(转换), 它能够返回一个新的RDDs. Let’s start with a few actions:

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

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

现在让我们使用转换操作.我们将使用 filter 转换操作返回一个新的RDD包含这个文件的子集.

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

我们能够通过链式编程把transformations 操作和 actions操作连接一起:

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

更多的RDD操作

RDD操作因子和转换因子可以用来进行更加复杂的计算.让我们寻找到需要的数据:

• Scala
• Python

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

第一个map一行整数数值,创建一个新的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: spark.RDD[(String, Int)] = spark.ShuffledAggregatedRDD@71f027b8

这里, 我们组合了 flatMap, map, 和reduceByKey 转换因子来计算文件中每个单词的数量通过一个RDD (String, Int) 对. 通过Shell统计单数的数量, 我们能够使用 collect 因子:

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来连接集群, 更多的描述请看 programming guide.

独立的应用

假如我希望开发一个独立的应用通过Spark API. 我们可以通过 Scala (with sbt), Java (with Maven), and Python来调用开发Spark API.

• Scala
• Java
• Python

现在我们创建一个简单的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的子类可能不能正确的工作.

这个程序仅仅是统计README文件中包含‘a’和‘b’的数量.注意你需要替换你的本地的YOUR_SPARK_HOME 环境变量. 不同于前面的Spark shell例子, 它需要初始化SparkContext, 我们需要初始化 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.1"

为了使 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

下一章

配置运行你的第一个Spark程序!

• 想深入的了解SparkAPI, 可以打开 Spark programming guide, 或者看“Programming Guides” 菜单了解其他组件.
• 如果想在集群上面运行引用, 可以点击 deployment overview了解.
• 最后, Spark包含几个简单的实例 (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

 转载自 并发编程网 - ifeve.com