本项目是 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目录并启动按照以下方式启动它:

ScalaPython ./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操作因子和转换因子可以用来进行更加复杂的计算.让我们寻找到需要的数据:

ScalaPython 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 数据集缓存起来:

ScalaPython 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.

ScalaJavaPython

现在我们创建一个简单的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