用Apache Spark进行大数据处理四

如何安装Spark

安装和使用Spark有几种不同方式。你可以在自己的电脑上将Spark作为一个独立的框架安装或者从诸如Cloudera，HortonWorks或MapR之类的供应商处获取一个Spark虚拟机镜像直接使用。或者你也可以使用在云端环境（如Databricks
Cloud）安装并配置好的Spark。

在本文中，我们将把Spark作为一个独立的框架安装并在本地启动它。最近Spark刚刚发布了1.2.0版本。我们将用这一版本完成示例应用的代码展示。

如何运行Spark

当你在本地机器安装了Spark或使用了基于云端的Spark后，有几种不同的方式可以连接到Spark引擎。

下表展示了不同的Spark运行模式所需的Master URL参数。

 

如何与Spark交互

Spark启动并运行后，可以用Spark shell连接到Spark引擎进行交互式数据分析。Spark
shell支持Scala和Python两种语言。Java不支持交互式的Shell，因此这一功能暂未在Java语言中实现。

可以用spark-shell.cmd和pyspark.cmd命令分别运行Scala版本和Python版本的Spark
Shell。

Spark网页控制台

不论Spark运行在哪一种模式下，都可以通过访问Spark网页控制台查看Spark的作业结果和其他的统计数据，控制台的URL地址如下：
http://localhost:4040
Spark控制台如下图3所示，包括Stages，Storage，Environment和Executors四个标签页

（点击查看大图）

 

图3. Spark网页控制台

共享变量

Spark提供两种类型的共享变量可以提升集群环境中的Spark程序运行效率。分别是广播变量和累加器。

广播变量：广播变量可以在每台机器上缓存只读变量而不需要为各个任务发送该变量的拷贝。他们可以让大的输入数据集的集群拷贝中的节点更加高效。

下面的代码片段展示了如何使用广播变量。

//
// Broadcast Variables
//
val broadcastVar = sc.broadcast(Array(1, 2, 3))
broadcastVar.value
累加器：只有在使用相关操作时才会添加累加器，因此它可以很好地支持并行。累加器可用于实现计数（就像在MapReduce中那样）或求和。可以用add方法将运行在集群上的任务添加到一个累加器变量中。不过这些任务无法读取变量的值。只有驱动程序才能够读取累加器的值。

下面的代码片段展示了如何使用累加器共享变量：

//
// Accumulators
//
 
val accum = sc.accumulator(0, "My Accumulator")
 
sc.parallelize(Array(1, 2, 3, 4)).foreach(x => accum += x)
 
accum.value

Spark应用示例

本篇文章中所涉及的示例应用是一个简单的字数统计应用。这与学习用Hadoop进行大数据处理时的示例应用相同。我们将在一个文本文件上执行一些数据分析查询。本示例中的文本文件和数据集都很小，不过无须修改任何代码，示例中所用到的Spark查询同样可以用到大容量数据集之上。

为了让讨论尽量简单，我们将使用Spark Scala Shell。

首先让我们看一下如何在你自己的电脑上安装Spark。

前提条件：

· 为了让Spark能够在本机正常工作，你需要安装Java开发工具包（JDK）。这将包含在下面的第一步中。

· 同样还需要在电脑上安装Spark软件。下面的第二步将介绍如何完成这项工作。

注：下面这些指令都是以Windows环境为例。如果你使用不同的操作系统环境，需要相应的修改系统变量和目录路径已匹配你的环境。

I. 安装JDK

1）从Oracle网站上下载JDK。推荐使用JDK
1.7版本。

将JDK安装到一个没有空格的目录下。对于Windows用户，需要将JDK安装到像c:\dev这样的文件夹下，而不能安装到“c:\Program
Files”文件夹下。“c:\Program Files”文件夹的名字中包含空格，如果软件安装到这个文件夹下会导致一些问题。

注：不要在“c:\Program Files”文件夹中安装JDK或（第二步中所描述的）Spark软件。

2）完成JDK安装后，切换至JDK 1.7目录下的”bin“文件夹，然后键入如下命令，验证JDK是否正确安装：

java -version

如果JDK安装正确，上述命令将显示Java版本。

II. 安装Spark软件：

从Spark网站上下载最新版本的Spark。在本文发表时，最新的Spark版本是1.2。你可以根据Hadoop的版本选择一个特定的Spark版本安装。我下载了与Hadoop
2.4或更高版本匹配的Spark，文件名是spark-1.2.0-bin-hadoop2.4.tgz。

将安装文件解压到本地文件夹中（如：c:\dev）。

为了验证Spark安装的正确性，切换至Spark文件夹然后用如下命令启动Spark Shell。这是Windows环境下的命令。如果使用Linux或Mac
OS，请相应地编辑命令以便能够在相应的平台上正确运行。

c:
cd c:\dev\spark-1.2.0-bin-hadoop2.4
bin\spark-shell
如果Spark安装正确，就能够在控制台的输出中看到如下信息。

….
15/01/17 23:17:46 INFO HttpServer: Starting HTTP Server
15/01/17 23:17:46 INFO Utils: Successfully started service 'HTTP class server' on port 58132.
Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /___/ .__/\_,_/_/ /_/\_\   version 1.2.0
      /_/
 
Using Scala version 2.10.4 (Java HotSpot(TM) 64-Bit Server VM, Java 1.7.0_71)
Type in expressions to have them evaluated.
Type :help for more information.
….
15/01/17 23:17:53 INFO BlockManagerMaster: Registered BlockManager
15/01/17 23:17:53 INFO SparkILoop: Created spark context..
Spark context available as sc.
可以键入如下命令检查Spark Shell是否工作正常。

sc.version
（或）

sc.appName
完成上述步骤之后，可以键入如下命令退出Spark Shell窗口：

:quit
如果想启动Spark Python Shell，需要先在电脑上安装Python。你可以下载并安装Anaconda，这是一个免费的Python发行版本，其中包括了一些比较流行的科学、数学、工程和数据分析方面的Python包。

然后可以运行如下命令启动Spark Python Shell：

c:
cd c:\dev\spark-1.2.0-bin-hadoop2.4
bin\pyspark

Spark示例应用

完成Spark安装并启动后，就可以用Spark API执行数据分析查询了。

这些从文本文件中读取并处理数据的命令都很简单。我们将在这一系列文章的后续文章中向大家介绍更高级的Spark框架使用的用例。

首先让我们用Spark API运行流行的Word Count示例。如果还没有运行Spark Scala Shell，首先打开一个Scala
Shell窗口。这个示例的相关命令如下所示：

import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
 
val txtFile = "README.md"
val txtData = sc.textFile(txtFile)
txtData.cache()
我们可以调用cache函数将上一步生成的RDD对象保存到缓存中，在此之后Spark就不需要在每次数据查询时都重新计算。需要注意的是，cache()是一个延迟操作。在我们调用cache时，Spark并不会马上将数据存储到内存中。只有当在某个RDD上调用一个行动时，才会真正执行这个操作。

现在，我们可以调用count函数，看一下在文本文件中有多少行数据。

txtData.count()
然后，我们可以执行如下命令进行字数统计。在文本文件中统计数据会显示在每个单词的后面。

val wcData = txtData.flatMap(l => l.split(" ")).map(word => (word, 1)).reduceByKey(_ + _)
 
wcData.collect().foreach(println)
如果想查看更多关于如何使用Spark核心API的代码示例，请参考网站上的Spark文档。

后续计划

在后续的系列文章中，我们将从Spark SQL开始，学习更多关于Spark生态系统的其他部分。之后，我们将继续了解Spark Streaming，Spark
MLlib和Spark GraphX。我们也会有机会学习像Tachyon和BlinkDB等框架。

小结

在本文中，我们了解了Apache Spark框架如何通过其标准API帮助完成大数据处理和分析工作。我们还对Spark和传统的MapReduce实现（如Apache
Hadoop）进行了比较。Spark与Hadoop基于相同的HDFS文件存储系统，因此如果你已经在Hadoop上进行了大量投资和基础设施建设，可以一起使用Spark和MapReduce。

此外，也可以将Spark处理与Spark SQL、机器学习以及Spark Streaming结合在一起。关于这方面的内容我们将在后续的文章中介绍。

利用Spark的一些集成功能和适配器，我们可以将其他技术与Spark结合在一起。其中一个案例就是将Spark、Kafka和Apache
Cassandra结合在一起，其中Kafka负责输入的流式数据，Spark完成计算，最后Cassandra
NoSQL数据库用于保存计算结果数据。

不过需要牢记的是，Spark生态系统仍不成熟，在安全和与BI工具集成等领域仍然需要进一步的改进。