第1章Spark概述

spark的产生背景

spark是如何产生的，这要先送大数据说起，大数据是如何产生的？Google就是处理大数据的，网页和网页之间有很多的关联关系，为了处理排序啊这些算法，所以Google就发明了，Google就发布了三个论文，基于这三个论文的开源，实现了Hadoop、Hdfs、MapReduce、Hbase等，但是感觉好像每次MapReduce只能处理一次数据，而且开始跑mapreduce的时候，时间有点长，做完一个提交一个，中间的这些处理其实没有迭代性，虽然有些工具能够帮你去处理迭代，比如说有的是做一个业务流程的开发，可以连续的调用几个MapReduce写的jar包，但是流程还是很长，所有的文件都是存在磁盘上面的，这样就会导致两个问题：

第一个就是迭代式计算

就好比轻功，这个脚踩着这个脚，迭代的往前走，就是不断有新的计算产生，然后新的计算使用前边计算的结果，然后新的产生的结果后边还需要用到这个结果，然后一直迭代下去。

比如机器学习推化问题的时候，这个迭代就显得非常重要了。

显然MapReduce在迭代计算上是比较弱的，很多原因是没有把一系列的步骤都写到一个MapReduce里面，MapReduce只有两个，一个是map映射，一个是reduce预约

第二个交互式的数据挖掘

hadoop无法做交互式的数据挖掘，我输入进去，立马展示出来，而不能说输入东西了，等了一个小时，然后交互一次，然后再输入，再交互，这就不叫交互了，等的时间太长，

hadoop在这两方面是不擅长的，更擅长于长时间的大数据量的计算，所以跟据reduce的分布式模型，就产生了spark。

上图中hadoop是先写入磁盘，然后计算，然后再写入磁盘然后再计算，先不说磁盘，就中间提交任务，任务与任务之间没有任何关联，单纯的是调度层面的关联，那么这个东西其实不是一个整体的东西

上图中的spark，只需要读取一次数据，而且在这一系列的数据中是在一个编程模型里面完成的，也就是说在这一个编程模型里面完成这些东西，只需要提交一次，然后里面直接可以进行计算，而且任务与任务之间时是可以进行内存数据交换的

从这两个模型看，spark对迭代计算和交互式数据挖掘提供了很强的支持。

1.1什么是Spark

官网：http://spark.apache.org

Spark是一种快速、通用、可扩展的大数据分析引擎，2009年诞生于加州大学伯克利分校AMPLab，2010年开源，2013年6月成为Apache孵化项目，2014年2月成为Apache顶级项目。项目是用Scala进行编写。

目前，Spark生态系统已经发展成为一个包含多个子项目的集合，其中包含SparkSQL、Spark Streaming、GraphX、MLib、SparkR等子项目，Spark是基于内存计算的大数据并行计算框架。除了扩展了广泛使用的 MapReduce 计算模型，而且高效地支持更多计算模式，包括交互式查询和流处理。Spark 适用于各种各样原先需要多种不同的分布式平台的场景，包括批处理、迭代算法、交互式查询、流处理。通过在一个统一的框架下支持这些不同的计算，Spark 使我们可以简单而低耗地把各种处理流程整合在一起。而这样的组合，在实际的数据分析过程中是很有意义的。不仅如此，Spark 的这种特性还大大减轻了原先需要对各种平台分别管理的负担。

大一统的软件栈，各个组件关系密切并且可以相互调用，这种设计有几个好处：1、软件栈中所有的程序库和高级组件都可以从下层的改进中获益。2、运行整个软件栈的代价变小了。不需要运行 5 到 10 套独立的软件系统了，一个机构只需要运行一套软件系统即可。系统的部署、维护、测试、支持等大大缩减。3、能够构建出无缝整合不同处理模型的应用。

Spark的内置项目如下：

1、软件栈中所有的程序库和高级组件都可以从下层的改进中获益。

spark这种all in one的方式，底层加快了速度，上层的速度也会跟着加快，因为是基于底层运行的

2、运行整个软件栈的代价变小了。不需要运行5到10套独立的软件系统了，一-个机构只需要运行一套软件系统即可。系统的部署、维护、测试、支持等大大缩减。

如上把sql、streaming、MLib、Graghx集成在一块

3、能够构建出无缝整合不同处理模型的应用。

如上sql、streaming、MLib、Graghx都是在spark core的基础上的，那么他们的抽象其实都是想通的，他们之间可以有数据交互，也就是提供了”1+1>=2“（这句话我也不知道啥意思，后面可能就理解了把）

Spark Core：实现了 Spark 的基本功能，包含任务调度、内存管理、错误恢复、与存储系统交互等模块。Spark Core 中还包含了对弹性分布式数据集(resilient distributed dataset，简称RDD)的 API 定义。

Spark SQL：是 Spark 用来操作结构化数据的程序包。通过 Spark SQL，我们可以使用 SQL 或者 Apache Hive 版本的 SQL 方言(HQL)来查询数据。Spark SQL 支持多种数据源，比如 Hive 表、Parquet 以及 JSON 等。

Spark Streaming：是 Spark 提供的对实时数据进行流式计算的组件。提供了用来操作数据流的 API，并且与 Spark Core 中的 RDD API 高度对应。

Spark MLlib：提供常见的机器学习(ML)功能的程序库。包括分类、回归、聚类、协同过滤等，还提供了模型评估、数据导入等额外的支持功能。

集群管理器：Spark 设计为可以高效地在一个计算节点到数千个计算节点之间伸缩计算。为了实现这样的要求，同时获得最大灵活性，Spark 支持在各种集群管理器(cluster manager)上运行，包括 Hadoop YARN、Apache Mesos，以及 Spark 自带的一个简易调度器，叫作独立调度器。

Spark得到了众多大数据公司的支持，这些公司包括Hortonworks、IBM、Intel、Cloudera、MapR、Pivotal、百度、阿里、腾讯、京东、携程、优酷土豆。当前百度的Spark已应用于凤巢、大搜索、直达号、百度大数据等业务；阿里利用GraphX构建了大规模的图计算和图挖掘系统，实现了很多生产系统的推荐算法；腾讯Spark集群达到8000台的规模，是当前已知的世界上最大的Spark集群。

1.2Spark特点

快

与Hadoop的MapReduce相比，Spark基于内存的运算要快100倍以上，基于硬盘的运算也要快10倍以上。Spark实现了高效的DAG执行引擎，可以通过基于内存来高效处理数据流。计算的中间结果是存在于内存中的。

易用

Spark支持Java、Python和Scala的API，还支持超过80种高级算法，使用户可以快速构建不同的应用。而且Spark支持交互式的Python和Scala的shell，可以非常方便地在这些shell中使用Spark集群来验证解决问题的方法。

通用

Spark提供了统一的解决方案。Spark可以用于批处理、交互式查询（Spark SQL）、实时流处理（Spark Streaming）、机器学习（Spark MLlib）和图计算（GraphX）。这些不同类型的处理都可以在同一个应用中无缝使用。Spark统一的解决方案非常具有吸引力，毕竟任何公司都想用统一的平台去处理遇到的问题，减少开发和维护的人力成本和部署平台的物力成本。

兼容性

Spark可以非常方便地与其他的开源产品进行融合。比如，Spark可以使用Hadoop的YARN和Apache Mesos作为它的资源管理和调度器，器，并且可以处理所有Hadoop支持的数据，包括HDFS、HBase和Cassandra等。这对于已经部署Hadoop集群的用户特别重要，因为不需要做任何数据迁移就可以使用Spark的强大处理能力。Spark也可以不依赖于第三方的资源管理和调度器，它实现了Standalone作为其内置的资源管理和调度框架，这样进一步降低了Spark的使用门槛，使得所有人都可以非常容易地部署和使用Spark。此外，Spark还提供了在EC2上部署Standalone的Spark集群的工具。

1.3Spark的用户和用途

我们大致把Spark的用例分为两类：数据科学应用和数据处理应用。也就对应的有两种人群：数据科学家和工程师。

数据科学任务

主要是数据分析领域，数据科学家要负责分析数据并建模，具备 SQL、统计、预测建模(机器学习)等方面的经验，以及一定的使用 Python、 Matlab 或 R 语言进行编程的能力。

数据处理应用

工程师定义为使用 Spark 开发生产环境中的数据处理应用的软件开发者，通过对接Spark的API实现对处理的处理和转换等任务。

第2章Spark集群安装

一共支持的有 local、local-cluster、standalone、Yarn、Mesos

spark目前集群安装支持三种类型，一种叫做独立调度器，standalone模式，也就是他自己不需要外部的任何支撑组件，就可以部署，在standalone中可以见到master 和 worker这两个角色，除此之外还支持YARN和Mesos这两种调度，这两个的职责是资源的划分，yarn前面学习过了，mesos就是当前有一个采集器，在这个采集器中又运行了hadoop，又运行了storm，每一个大数据的框架都有自己的一个资源调度的机制，软件与软件之间并没有一个资源协调机制，这就会出现资源争抢，然后把其他框架的内存都吃满了，这个时候非常需要一个资源的协调者，但是需要注意的是，如果用的是mesos这个框架，那么在部署的时候master和worker这两个角色就被干掉了，master和worker存在的意义就是standalone

这里是standalone的方式

2.1集群角色

从物理部署层面上来看，Spark主要分为两种类型的节点，Master节点和Worker节点，Master节点主要运行集群管理器的中心化部分，所承载的作用是分配Application到Worker节点，维护Worker节点，Driver，Application的状态。Worker节点负责具体的业务运行。

从Spark程序运行的层面来看，Spark主要分为驱动器节点和执行器节点。

上图中的cluster manager就是master

2.2机器准备

准备两台以上Linux服务器，安装好JDK1.8

这里准备了3台

2.3下载Spark安装包

上传解压安装包

上传spark-2.1.1-bin-hadoop2.7.tgz安装包到Linux上

解压安装包到指定位置

tar -xf spark-2.1.1-bin-hadoop2.7.tgz -C /home/bigdata/Hadoop

2.4配置Spark

Spark的部署模式有Local、Local-Cluster、Standalone、Yarn、Mesos，我们选择最具代表性的Standalone集群部署模式。

进入到Spark安装目录

cd /home/bigdata/hadoop/spark-2.1.1-bin-hadoop2.7/conf

将slaves.template复制为slaves

将spark-env.sh.template复制为spark-env.sh

修改slave文件，将work的hostname输入：

如上图 slave01 slave02 ，这里原来是localhost

修改spark-env.sh文件，添加如下配置：

spark master的ip 和端口

将配置好的Spark文件拷贝到其他节点上

scp slaves bigdata@slave01:~/hadoop/spark-2.1.1-bin-hadoop2.7/conf

scp slaves bigdata@slave02:~/hadoop/spark-2.1.1-bin-hadoop2.7/conf

scp spark-env.sh bigdata@slave01:~/hadoop/spark-2.1.1-bin-hadoop2.7/conf

scp spark-env.sh bigdata@slave02:~/hadoop/spark-2.1.1-bin-hadoop2.7/conf

Spark集群配置完毕，目前是1个Master，2个Work，master01上启动Spark集群

/home/bigdata/hadoop/spark-2.1.1-bin-hadoop2.7/sbin/start-all.sh

启动后执行jps命令，主节点上有Master进程，其他子节点上有Work进行，登录Spark管理界面查看集群状态（主节点）：http://master01:8080/

到此为止，Spark集群安装完毕.

注意：如果遇到 “JAVA_HOME not set” 异常，可以在sbin目录下的spark-config.sh 文件中加入如下配置：

export JAVA_HOME=XXXX

如果遇到Hadoop HDFS的写入权限问题：

org.apache.hadoop.security.AccessControlException

解决方案：在hdfs-site.xml中添加如下配置，关闭权限验证

 &lt;property&gt;
 &lt;name&gt;dfs.permissions&lt;/name&gt;
 &lt;value&gt;false&lt;/value&gt;
 &lt;/property&gt;

<name>dfs.permissions</name>

<value>false</value>

</property>

2.5配置Job History Server

进入到Spark安装目录

cd /home/bigdata/hadoop/spark-2.1.1-bin-hadoop2.7/con

将spark-default.conf.template复制为spark-default.conf

修改spark-default.conf文件，开启Log：

修改如上的白字的部分，spark相关的ip port 改为自己的

修改spark-env.sh文件，添加如下配置：

在HDFS上创建好你所指定的eventLog日志目录。

spark-defaults.conf
spark.eventLog.enabled true
spark.eventLog.dir 
hdfs://master01:9000/directory
spark.eventLog.compress true

spark-defaults.conf

spark.eventLog.enabled true

spark.eventLog.dir

hdfs://master01:9000/directory

spark.eventLog.compress true

spark-env.sh

export SPARK_HISTORY_OPTS="-Dspark.history.ui.port=7077 
-Dspark.history.retainedApplications=3
-Dspark.history.fs.logDirectory=hdfs://master01:9000/directory"

export SPARK_HISTORY_OPTS="-Dspark.history.ui.port=7077

-Dspark.history.retainedApplications=3

-Dspark.history.fs.logDirectory=hdfs://master01:9000/directory"

参数描述：

spark.eventLog.dir：Application在运行过程中所有的信息均记录在该属性指定的路径下；

spark.history.ui.port=4000 调整WEBUI访问的端口号为4000 访问history server时候的端口，就是上面配置history server的时候那个7077

spark.history.retainedApplications=3 指定保存Application历史记录的个数，如果超过这个值，旧的应用程序信息将被删除，这个是内存中的应用数，而不是页面上显示的应用数。保存在内存里面保存三个job的历史信息

spark.history.fs.logDirectory=hdfs://master01:9000/directory 配置了该属性后，在start-history-server.sh时就无需再显式的指定路径，Spark History Server页面只展示该指定路径下的信息，把上面配置的spark.history.retainedApplications=3保存的信息扔到hdfs里面

将配置好的Spark文件拷贝到其他节点上

/home/bigdata/hadoop/spark-2.1.1-bin-hadoop2.7/sbin/start-all.sh

启动相应的HDFS

启动后执行

/home/bigdata/hadoop/spark-2.1.1-bin-hadoop2.7/sbin/start-history-server.sh

到此为止，Spark History Server安装完毕.

2.6配置Spark HA

集群部署完了，但是有一个很大的问题，那就是Master节点存在单点故障，要解决此问题，就要借助zookeeper，并且启动至少两个Master节点来实现高可靠，配置方式比较简单：

Spark集群规划：master01，master02是Master；slave01，slave02，slave03是Worker

安装配置Zookeeper集群，并启动Zookeeper集群

停止spark所有服务，修改配置文件spark-env.sh，在该配置文件中删掉SPARK_MASTER_IP并添加如下配置

export SPARK_DAEMON_JAVA_OPTS="
 -Dspark.deploy.recoveryMode=ZOOKEEPER
 -Dspark.deploy.zookeeper.url=zk1,zk2,zk3
 -Dspark.deploy.zookeeper.dir=/spark"

export SPARK_DAEMON_JAVA_OPTS="

-Dspark.deploy.recoveryMode=ZOOKEEPER

-Dspark.deploy.zookeeper.url=zk1,zk2,zk3

-Dspark.deploy.zookeeper.dir=/spark"

1.在 master01节点上修改slaves配置文件内容指定worker节点

2.将配置文件同步到所有节点。

3.在master01上执行sbin/start-all.sh脚本，启动集群并启动第一个master节点，然后在master02上执行sbin/start-master.sh启动第二个master节点。

4.程序中spark集群的访问地址需要改成：

spark://master01:port1,master02:port2

第3章执行Spark程序

3.1执行第一个spark程序

直接执行下面的指令就可以，这里用的是spark的一个小例子

/home/bigdata/hadoop/spark-2.1.1-bin-hadoop2.7/bin/spark-submit \
--class org.apache.spark.examples.SparkPi \
--master spark://master01:7077 \
--executor-memory 1G \
--total-executor-cores 2 \
/home/bigdata/hadoop/spark-2.1.1-bin-hadoop2.7/examples/jars/spark-examples_2.11-2.1.1.jar \
100

/home/bigdata/hadoop/spark-2.1.1-bin-hadoop2.7/bin/spark-submit \

--class org.apache.spark.examples.SparkPi \

--master spark://master01:7077 \

--executor-memory 1G \

--total-executor-cores 2 \

/home/bigdata/hadoop/spark-2.1.1-bin-hadoop2.7/examples/jars/spark-examples_2.11-2.1.1.jar \

100

参数说明：

–class 主类

–master spark://master01:7077 指定Master的地址

–executor-memory 1G 指定每个executor可用内存为1G

–total-executor-cores 2 指定每个executor使用的cup核数为2个

该算法是利用蒙特·卡罗算法求PI

为什么上面一个日志那么多，一个日志那么少？

需要修改conf 下的log4j.properties.template这个文件，如上图，复制一份，名字叫做log4j.properties，然后开始修改log4j.properties.

这里原来的日志级别是info，应该改为warn

3.2Spark应用提交

一旦打包好,就可以使用bin/spark-submit脚本启动应用了. 这个脚本负责设置spark使用的classpath和依赖,支持不同类型的集群管理器和发布模式:

./bin/spark-submit

–class

–master

–deploy-mode

–conf =

… # other options

[application-arguments]

一些常用选项:

1)–class: 你的应用的启动类 (如 org.apache.spark.examples.SparkPi)

2)–master: 集群的master URL (如 spark://23.195.26.187:7077)

3)–deploy-mode: 是否发布你的驱动到worker节点(cluster) 或者作为一个本地客户端 (client) (default: client)*

4)–conf: 任意的Spark配置属性，格式key=value. 如果值包含空格，可以加引号“key=value”. 缺省的Spark配置

5)application-jar: 打包好的应用jar,包含依赖. 这个URL在集群中全局可见。比如hdfs:// 共享存储系统，如果是 file:// path，那么所有的节点的path都包含同样的jar.

6)application-arguments: 传给main()方法的参数

local本地以一个worker线程运行(例如非并行的情况).

local[K]本地以K worker 线程 (理想情况下, K设置为你机器的CPU核数).

local[*]本地以本机同样核数的线程运行.

spark://HOST:PORT连接到指定的Spark standalone cluster master. 端口是你的master集群配置的端口，缺省值为7077.

mesos://HOST:PORT连接到指定的Mesos 集群. Port是你配置的mesos端口，缺省是5050. 或者如果Mesos使用ZOoKeeper,格式为 mesos://zk://…

yarn-client以client模式连接到YARN cluster. 集群的位置基于HADOOP_CONF_DIR 变量找到.

yarn-cluster以cluster模式连接到YARN cluster. 集群的位置基于HADOOP_CONF_DIR 变量找到.

Master URL 可以是以下格式：

查看Spark-submit全部参数：

3.3启动Spark Shell

spark-shell是Spark自带的交互式Shell程序，方便用户进行交互式编程，用户可以在该命令行下用scala编写spark程序。

3.3.1启动Spark shell

/home/bigdata/hadoop/spark-2.1.1-bin-hadoop2.7/bin/spark-shell

–master spark://master01:7077

–executor-memory 2g

–total-executor-cores 2

注意：

如果启动spark shell时没有指定master地址，但是也可以正常启动spark shell和执行spark shell中的程序，其实是启动了spark的cluster模式，如果spark是单节点，并且没有指定slave文件，这个时候如果打开spark-shell 默认是local模式

Local模式是master和worker在同同一进程内

Cluster模式是master和worker在不同进程内

Spark Shell中已经默认将SparkContext类初始化为对象sc。用户代码如果需要用到，则直接应用sc即可

3.3.2在Spark shell中编写WordCount程序

首先启动hdfs

将Spark目录下的RELEASE文件上传一个文件到hdfs://master01:9000/RELEASE

~/hadoop/hadoop-2.7.3/bin/hdfs dfs -put ./RELEASE /

在Spark shell中用scala语言编写spark程序

spark中提供了spark shell 命令

bin/spark-shell

1	bin/spark-shell

为什么会出现这个warning呢？

是因为如果希望消除这个warning，需要把hadoop重新在机器上编译一下，他有一些本地包，把这些本地包，替换到spark目录下面。

执行：

sc.textFile("hdfs://master01:9000/RELEASE").flatMap(_.split(" "))
.map((_,1)).reduceByKey(_+_).saveAsTextFile("hdfs://master01:9000/out")

1 2	sc.textFile("hdfs://master01:9000/RELEASE").flatMap(_.split(" ")) .map((_,1)).reduceByKey(_+_).saveAsTextFile("hdfs://master01:9000/out")

为什么是sc呢？从下面的日志中也能够看到 apark context web UI available as ‘sc’，apark context 就是进入整个spark体系的入口，所以所有的程序都要从sc开始

textFile()获取文件

flatMap(.split(” “) 获取的是一行一行的，通过””将每一行的分隔开成单词

.map((,1))将每个word转化成一个元组，

reduceByKey(+) 按照key 即word ，把相同word的放在一起

.saveAsTextFile(“hdfs://master01:9000/out” 加在一起之后把结果放在hdfs上

使用hdfs命令查看结果

hdfs dfs -cat hdfs://master01:9000/out/p*

说明：

sc是SparkContext对象，该对象时提交spark程序的入口

textFile(hdfs://master01:9000/RELEASE)是hdfs中读取数据

flatMap(.split(” “))先map在压平

map((,1))将单词和1构成元组

reduceByKey(+)按照key进行reduce，并将value累加

saveAsTextFile(“hdfs:// master01:9000/out”)将结果写入到hdfs中

以上可以清晰的感觉到spark和MapReduce的区别，哎舒服

3.4在IDEA中编写WordCount程序

spark shell仅在测试和验证我们的程序时使用的较多，在生产环境中，通常会在IDE中编制程序，然后打成jar包，然后提交到集群，最常用的是创建一个Maven项目，利用Maven来管理jar包的依赖。

1.创建一个项目

2.选择Maven项目，然后点击next

3.填写maven的GAV，然后点击next

4.填写项目名称，然后点击finish

5.创建好maven项目后，点击Enable Auto-Import

6.配置Maven的pom.xml

&lt;?xml version="1.0" encoding="UTF-8"?&gt;
&lt;project xmlns="http://maven.apache.org/POM/4.0.0"
 xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
 xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"&gt;
 &lt;parent&gt;
 &lt;artifactId&gt;spark&lt;/artifactId&gt;
 &lt;groupId&gt;com.atguigu&lt;/groupId&gt;
 &lt;version&gt;1.0-SNAPSHOT&lt;/version&gt;
 &lt;/parent&gt;
 &lt;modelVersion&gt;4.0.0&lt;/modelVersion&gt;
 &lt;artifactId&gt;wordcount&lt;/artifactId&gt;
 &lt;dependencies&gt;
 //scala
 &lt;dependency&gt;
 &lt;groupId&gt;org.scala-lang&lt;/groupId&gt;
 &lt;artifactId&gt;scala-library&lt;/artifactId&gt;
 &lt;version&gt;${scala.version}&lt;/version&gt;
 &lt;scope&gt;provided&lt;/scope&gt;
 &lt;/dependency&gt;
//spark
 &lt;dependency&gt;
 &lt;groupId&gt;org.apache.spark&lt;/groupId&gt;
 &lt;artifactId&gt;spark-core_2.11&lt;/artifactId&gt;
 &lt;version&gt;${spark.version}&lt;/version&gt;
 &lt;scope&gt;provided&lt;/scope&gt;
 &lt;/dependency&gt;
 &lt;dependency&gt;
 &lt;groupId&gt;org.apache.hadoop&lt;/groupId&gt;
 &lt;artifactId&gt;hadoop-client&lt;/artifactId&gt;
 &lt;version&gt;${hadoop.version}&lt;/version&gt;
 &lt;scope&gt;provided&lt;/scope&gt;
 &lt;/dependency&gt;
 &lt;!-- Logging --&gt;
 &lt;dependency&gt;
 &lt;groupId&gt;org.slf4j&lt;/groupId&gt;
 &lt;artifactId&gt;jcl-over-slf4j&lt;/artifactId&gt;
 &lt;version&gt;${slf4j.version}&lt;/version&gt;
 &lt;/dependency&gt;
 &lt;dependency&gt;
 &lt;groupId&gt;org.slf4j&lt;/groupId&gt;
 &lt;artifactId&gt;slf4j-api&lt;/artifactId&gt;
 &lt;version&gt;${slf4j.version}&lt;/version&gt;
 &lt;/dependency&gt;
 &lt;dependency&gt;
 &lt;groupId&gt;org.slf4j&lt;/groupId&gt;
 &lt;artifactId&gt;slf4j-log4j12&lt;/artifactId&gt;
 &lt;version&gt;${slf4j.version}&lt;/version&gt;
 &lt;/dependency&gt;
 &lt;dependency&gt;
 &lt;groupId&gt;log4j&lt;/groupId&gt;
 &lt;artifactId&gt;log4j&lt;/artifactId&gt;
 &lt;version&gt;${log4j.version}&lt;/version&gt;
 &lt;/dependency&gt;
 &lt;!-- Logging End --&gt;
 &lt;/dependencies&gt;
 &lt;build&gt;
 &lt;finalName&gt;wordcount&lt;/finalName&gt;
 &lt;plugins&gt;
 &lt;plugin&gt;
 &lt;groupId&gt;net.alchim31.maven&lt;/groupId&gt;
 &lt;artifactId&gt;scala-maven-plugin&lt;/artifactId&gt;
 &lt;version&gt;3.2.2&lt;/version&gt;
 &lt;executions&gt;
 &lt;execution&gt;
 &lt;goals&gt;
 &lt;goal&gt;compile&lt;/goal&gt;
 &lt;goal&gt;testCompile&lt;/goal&gt;
 &lt;/goals&gt;
 &lt;/execution&gt;
 &lt;/executions&gt;
 &lt;/plugin&gt;
 &lt;plugin&gt;
 &lt;groupId&gt;org.apache.maven.plugins&lt;/groupId&gt;
 &lt;artifactId&gt;maven-assembly-plugin&lt;/artifactId&gt;
 &lt;version&gt;3.0.0&lt;/version&gt;
 &lt;configuration&gt;
 &lt;archive&gt;
 &lt;manifest&gt;
 &lt;mainClass&gt;com.atguigu.spark.WordCount&lt;/mainClass&gt;
 &lt;/manifest&gt;
 &lt;/archive&gt;
 &lt;descriptorRefs&gt;
 &lt;descriptorRef&gt;jar-with-dependencies&lt;/descriptorRef&gt;
 &lt;/descriptorRefs&gt;
 &lt;/configuration&gt;
 &lt;executions&gt;
 &lt;execution&gt;
 &lt;id&gt;make-assembly&lt;/id&gt;
 &lt;phase&gt;package&lt;/phase&gt;
 &lt;goals&gt;
 &lt;goal&gt;single&lt;/goal&gt;
 &lt;/goals&gt;
 &lt;/execution&gt;
 &lt;/executions&gt;
 &lt;/plugin&gt;
 &lt;/plugins&gt;
 &lt;/build&gt;
&lt;/project&gt;

<?xml version="1.0" encoding="UTF-8"?>

<project xmlns="http://maven.apache.org/POM/4.0.0"

xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">

<artifactId>spark</artifactId>

<groupId>com.atguigu</groupId>

<version>1.0-SNAPSHOT</version>

</parent>

<artifactId>wordcount</artifactId>

//scala

<groupId>org.scala-lang</groupId>

<artifactId>scala-library</artifactId>

<version>${scala.version}</version>

<scope>provided</scope>

</dependency>

//spark

<groupId>org.apache.spark</groupId>

<artifactId>spark-core_2.11</artifactId>

<version>${spark.version}</version>

<scope>provided</scope>

</dependency>

<groupId>org.apache.hadoop</groupId>

<artifactId>hadoop-client</artifactId>

<version>${hadoop.version}</version>

<scope>provided</scope>

</dependency>

<version>${slf4j.version}</version>

</dependency>

<version>${slf4j.version}</version>

</dependency>

<version>${slf4j.version}</version>

</dependency>

<version>${log4j.version}</version>

</dependency>

</dependencies>

<build>

<finalName>wordcount</finalName>

<groupId>net.alchim31.maven</groupId>

<artifactId>scala-maven-plugin</artifactId>

<goals>

<goal>compile</goal>

<goal>testCompile</goal>

</goals>

</execution>

</executions>

</plugin>

<groupId>org.apache.maven.plugins</groupId>

<artifactId>maven-assembly-plugin</artifactId>

<mainClass>com.atguigu.spark.WordCount</mainClass>

</manifest>

</archive>

<descriptorRef>jar-with-dependencies</descriptorRef>

</descriptorRefs>

</configuration>

<id>make-assembly</id>

<phase>package</phase>

<goals>

<goal>single</goal>

</goals>

</execution>

</executions>

</plugin>

</plugins>

</build>

</project>

7.将src/main/scala设置成源代码目录。

8.添加IDEA Scala（执行此操作后，pom文件中不用添加scala依赖，应为已经以lib库的方式加入）

9.新建一个Scala class，类型为Object

10.编写spark程序

package com.atguigu.spark
import org.apache.spark.{SparkConf, SparkContext}
import org.slf4j.LoggerFactory
/**
 * Created by wuyufei on 31/07/2017.
 */
object WordCount {
 val logger = LoggerFactory.getLogger(WordCount.getClass)
 def main(args: Array[String]) {
 //创建SparkConf()并设置App名称
 val conf = new SparkConf().setAppName("WC")
 //创建SparkContext，该对象是提交spark App的入口
 val sc = new SparkContext(conf)
 //使用sc创建RDD并执行相应的transformation和action
 sc.textFile(args(0)).flatMap(_.split(" ")).map((_, 1)).reduceByKey(_+_, 1).sortBy(_._2, false).saveAsTextFile(args(1)) 
 //停止sc，结束该任务
 logger.info("complete!")
 sc.stop()
 }
}

package com.atguigu.spark

import org.apache.spark.{SparkConf, SparkContext}

import org.slf4j.LoggerFactory

/**

* Created by wuyufei on 31/07/2017.

object WordCount {

val logger = LoggerFactory.getLogger(WordCount.getClass)

def main(args: Array[String]) {

//创建SparkConf()并设置App名称

val conf = new SparkConf().setAppName("WC")

//创建SparkContext，该对象是提交spark App的入口

val sc = new SparkContext(conf)

//使用sc创建RDD并执行相应的transformation和action

sc.textFile(args(0)).flatMap(_.split(" ")).map((_, 1)).reduceByKey(_+_, 1).sortBy(_._2, false).saveAsTextFile(args(1))

//停止sc，结束该任务

logger.info("complete!")

sc.stop()

}

分步

11.使用Maven打包：首先修改pom.xml中的main class

12.点击idea右侧的Maven Project选项，点击Lifecycle,选择clean和package，然后点击Run Maven Build

13.选择编译成功的jar包，并将该jar上传到Spark集群中的某个节点上

生成了两个jar，我们需要的是with-dependencies这个jar

14.首先启动hdfs和Spark集群

启动hdfs

/home/bigdata/hadoop/hadoop-2.7.3/sbin/start-dfs.sh

启动spark

/home/bigdata/hadoop/spark-2.1.1-bin-hadoop2.7/sbin/start-all.sh

15.使用spark-submit命令提交Spark应用（注意参数的顺序）

/home/bigdata/hadoop/spark-2.1.1-bin-hadoop2.7/bin/spark-submit\
 --class com.atguigu.spark.WordCount\
 --master spark://master01:7077\
 --executor-memory 1G \
 --total-executor-cores 2 \
 wordcount-jar-with-dependencies.jar\
 hdfs://master01:9000/RELEASE\
 hdfs://master01:9000/out