本教程快速介绍了Spark的使用。 首先我们介绍了通过Spark 交互式shell调用API( Python或者scala代码)，然后演示如何使用Java, Scala或者Python编写独立程序。 你可以查看Spark编程指南了解完整的参考。

开始下面的快速入门之前，首先需要到 Spark网站 下载一份打包好的spark。 既然本教程中我们不使用HDFS,你可以随便下载一个适配任何Hadoop的版本的Spark。

本教程翻译时的Spark版本为1.1.1

使用Spark进行交互式分析

基本操作

Spark shell提供了一个简单方式去学习API，它也是一个交互式分析数据的强大工具。 你既可以使用Scala(运行在JVM之上，所以可以使用众多的Java库),也可以使用Python。运行Spark文件夹下的的命令:

1

. /bin/ spark-shell

Spark最主要的一个抽象出来的概念就是分布式的数据集合， 也就是弹性分布式数据集Resilient Distributed Dataset (RDD). RDD可以从Hadoop InputFormats (比如HDFS文件)创建， 也可以通过其它RDD转换(transforming)得到。 让我们从Spark源代码文件夹下的README文件创建一个RDD:

1

2

scala> val textFile = sc.textFile( "README.md" )

textFile: spark. RDD [ String ] = spark. MappedRDD @ 2 ee9b6e3

RDD包含 action ,可以返回数据， 也包含 transformation ,返回新的RDD的指针。 先看一些action的例子:

1

2

3

4

5

scala> textFile.count() // 此RDD中的item的数量

res0: Long = 126

scala> textFile.first() // 此RDD第一个item

res1: String = # Apache Spark

现在再看一个转换的例子。我们使用 filter 返回一个新的RDD， 新的RDD是文件中item的一个子集。

1

2

scala> val linesWithSpark = textFile.filter(line => line.contains( "Spark" ))

linesWithSpark: spark. RDD [ String ] = spark. FilteredRDD @ 7 dd4af09

将transformation和action串起来:

1

2

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

res3: Long = 15

更多的RDD操作

RDD的transformation和action可以组成起来完成复杂的计算。 比如查找包含最多单词的一行：

1

2

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

res4: Long = 15

第一步map一行包含的单词数到一个整数， 第二步调用reduce得到最大的单词数。map和reduce的参数都是lambda表达式(closures), 可以调用 Scala/Java库. 例如我们很容易的调用在其它地方声明的方法。 这里我们使用 Math.max() 函数简化代码：

1

2

3

4

5

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流很容易:

1

2

scala> val wordCounts = textFile.flatMap(line => line.split( " " )).map(word => (word, 1 )).reduceByKey((a, b) => a + b)

wordCounts: spark. RDD [( String , Int )] = spark. ShuffledAggregatedRDD @ 71 f027b8

此处我们使用flatMap, map 和 reduceByKey转换来计算文件中每个单词的频度。 为了收集单词频度结果，我们可以调用collect action:

1

2

scala> wordCounts.collect()

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

缓存

Spark也支持将数据集放入集群的内存中缓存起来. 当数据重复访问时特别有用， 比如查询一个小的 “hot”数据集或者运行一个交互式算法PageRank. 看一个简单的例子， 我们把上面的linesWithSpark数据集缓存起来:

1

2

3

4

5

6

7

8

scala> linesWithSpark.cache()

res7: spark. RDD [ String ] = spark. FilteredRDD @ 17e51082

scala> linesWithSpark.count()

res8: Long = 15

scala> linesWithSpark.count()

res9: Long = 15

当然使用Spark缓存一个100行的文本文件看起来有些傻，我们只是做个示范。 你可以将它用在非常大的数据集上，即使它们可能横跨几十甚至上百个节点。你也可以使用bin/spark-shell交互式实现此功能， 就像开发指南中描述的那样。

独立应用

下面我们想说一下怎样使用Spark API编写一个独立的应用程序。 这里使用Scala (SBT构建工具)和Java举例。 (Python官方文档中有，译者未翻译)

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

/* 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))

}

}

这个程序统计Spark README文件中包含字符 a 和 b 的行数。 注意你需要用你实际的Spark路径替换 YOUR_SPARK_HOME。 不像上面的Spark shell的例子， 我们初始化一个SparkContext 作为程序的一部分.

我们将一个SparkConf对象传给SparkContext的构造函数， 它包含了我们程序的信息。

我们的程序依赖Spark API,所以我们包含一个sbt配置文件：simple.sbt 指明Spark是一个依赖， 这个文件也增加了Spark依赖的仓库(repository):

1

2

3

4

name := "Simple Project"

version := "1.0"

scalaVersion := "2.10.4"

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

为了保证sbt工作正常，我们需要将SimpleApp.scala和simple.sbt放入典型的sbt项目布局的文件夹中。 如此一来我们将应用代码可以打包成一个jar文件， 然后使用spark-submit脚本来运行此程序。

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

# 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

或者使用Java

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

/* SimpleApp.java */

import org.apache.spark.api.java.*;

import org.apache.spark.SparkConf;

import org.apache.spark.api.java.function.Function;

public class SimpleApp {

public static void main (String[] args) {

String logFile = "YOUR_SPARK_HOME/README.md" ; // Should be some file on your system

SparkConf conf = new SparkConf().setAppName( "Simple Application" );

JavaSparkContext sc = new JavaSparkContext(conf);

JavaRDD logData = sc.textFile(logFile).cache();

long numAs = logData.filter( new Function() {

public Boolean call (String s) { return s.contains( "a" ); }

}).count();

long numBs = logData.filter( new Function() {

public Boolean call (String s) { return s.contains( "b" ); }

}).count();

System.out.println( "Lines with a: " + numAs + ", lines with b: " + numBs);

}

}

这个程序统计Spark README文件中包含字符 a 和 b 的行数。. 注意你需要用你实际的Spark路径替换 YOUR_SPARK_HOME。 不像上面的Spark shell的例子， 我们需要一个JavaSparkContext对象. 我们也创建了RDD (JavaRDD)然后运行transformations. 最后我们传递给Spark一个function对象， 这个function对象是一个匿名类，继承于 spark.api.java.function.Function. Spark开发指南描述了细节. (译者注： 这是Java 7的语法， 通过Java 8 Lambda表达式，上面的代码和scala一样的简化)

为了编译此程序，我们需要写一个Maven pom.xml文件， 增加Spark作为依赖. 注意Spark artifact带有Scala的版本.

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

< project >

< groupId > edu.berkeley groupId >

< artifactId > simple-project artifactId >

< modelVersion > 4.0.0 modelVersion >

< name > Simple Project name >

< packaging > jar packaging >

< version > 1.0 version >

< dependencies >

< dependency >

< groupId > org.apache.spark groupId >

< artifactId > spark-core_2.10 artifactId >

< version > 1.1.1 version >

dependency >

dependencies >

project >

使用Maven项目的布局:

1

2

3

4

5

6

$ find .

./pom.xml

./src

./src/main

./src/main/java

./src/main/java/SimpleApp.java

现在,我们使用Maven打包并使用./bin/spark-submit执行此程序.

1

2

3

4

5

6

7

8

9

10

11

12

# Package a jar containing your application

$ mvn package

...

[INFO] Building jar: {..}/{..}/target/simple-project- 1.0 .jar

# Use spark-submit to run your application

$ YOUR_SPARK_HOME/bin/spark-submit \

--class "SimpleApp" \

--master local [ 4 ] \

target/simple-project- 1.0 .jar

...

Lines with a: 46 , Lines with b: 23

深入了解

亲爱的读者，恭喜你运行了你的第一个Spark应用程序!你肯定不仅仅满足于此，以下是更多的深入学习的资料：

深度学习API和其它组件, 请参照 Spark开发指南

学习在集群中运行程序，访问 发布概览 .

最后, Spark发布包中的examples文件夹下包含几个例子 ( Scala , Java , Python ). 你可以运行它们:

1

2

3

4

5

# 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