Spark概述

什么是Spark（官网：http://spark.apache.org）

Spark是一种快速、通用、可扩展的大数据分析引擎。目前，Spark生态系统已经发展成为一个包含多个子项目的集合，其中包含SparkSQL、Spark Streaming、GraphX、MLlib等子项目，Spark是基于内存计算的大数据并行计算框架。Spark基于内存计算，提高了在大数据环境下数据处理的实时性，同时保证了高容错性和高可伸缩性，允许用户将Spark部署在大量廉价硬件之上，形成集群。

百度的Spark已应用于凤巢、大搜索、直达号、百度大数据等业务；
阿里利用GraphX构建了大规模的图计算和图挖掘系统，实现了很多生产系统的推荐算法；
腾讯Spark集群达到8000台的规模，是当前已知的世界上最大的Spark集群。

为什么要学Spark

中间结果输出：基于MapReduce的计算引擎通常会将中间结果输出到磁盘上，进行存储和容错。出于任务管道承接的，考虑，当一些查询翻译到MapReduce任务时，往往会产生多个Stage，而这些串联的Stage又依赖于底层文件系统（如HDFS）来存储每一个Stage的输出结果
Spark是MapReduce的替代方案，而且兼容HDFS、Hive，可融入Hadoop的生态系统，以弥补MapReduce的不足。

Spark特点

1、快

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

2、易用

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

3、用

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

4、兼容性

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

Spark任务调度

Spark 基础 - 图4

1.构建RDD

RDD Objects: 创建RDD对象，产生DAG（有向无环图）

2.DAG调度

DAGScheduler对作业的DAG图进行切分成不同的stage，每个stage都是任务的集合(taskset)并以taskset为单位提交(submitTasks)给TaskScheduler。

3.任务调度

TaskScheduler通过TaskSetManager管理任务(task)并通过集群中的资源管理器(Cluster Manager)把任务(task)发给集群中的Worker的Executor, 期间如果某个任务(task)失败， TaskScheduler会重试，TaskScheduler发现某个任务(task)一直未运行完成，有可能在不同机器启动同一个任务(task)，哪个任务(task)先运行完就用哪个任务(task)的结果。无论任务(task)运行成功或者失败TaskScheduler都会向DAGScheduler汇报当前状态，如果某个stage运行失败，TaskScheduler会通知DAGScheduler可能会重新提交任务。

4.任务执行

Worker接收到的是任务(task)，执行任务(task)的是进程中的线程（executor），一个进程中可以有多个线程工作进而可以处理多个数据分片(多线程方式执行),每一个线程负责一个任务，执行任务(task)、读取或存储数据。

Spark任务提交

Spark 基础 - 图5

Spark集群安装

安装

准备两台以上Linux服务器，安装好JDK1.7
上传解压安装包
上传spark-1.6.1-bin-hadoop2.6.tgz安装包到Linux上
解压安装包到指定位置
tar -zxvf spark-1.6.1-bin-hadoop2.6.tgz -C /root/apps

配置Spark

进入到Spark安装目录
cd /usr/local/spark-1.5.2-bin-hadoop2.6
进入conf目录并重命名并修改spark-env.sh.template文件
cd conf/
mv spark-env.sh.template spark-env.sh
vi spark-env.sh
在该配置文件中添加如下配置
export JAVA_HOME=/usr/java/jdk1.7.0_45
export SPARK_MASTER_IP=cdh1
export SPARK_MASTER_PORT=7077
保存退出
重命名并修改slaves.template文件
mv slaves.template slaves
vi slaves
在该文件中添加子节点所在的位置（Worker节点）
Mini1
Mini2
保存退出
将配置好的Spark拷贝到其他节点上
scp -r spark-1.5.2-bin-hadoop2.6/ Mini1:/root/apps
scp -r spark-1.5.2-bin-hadoop2.6/ Mini2:/root/apps

Spark集群配置完毕，目前是1个Master，3个Work，在cdh1上启动Spark集群
/root/apps/spark-1.5.2-bin-hadoop2.6/sbin/start-all.sh

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

Spark 基础 - 图6

到此为止，Spark集群安装完毕，但是有一个很大的问题，那就是Master节点存在单点故障，要解决此问题，就要借助zookeeper，并且启动至少两个Master节点来实现高可靠，配置方式比较简单：
Spark集群规划：node1，node2是Master；node3，node4，node5是Worker
安装配置zk集群，并启动zk集群
停止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”
1.在node1节点上修改slaves配置文件内容指定worker节点
2.在node1上执行sbin/start-all.sh脚本，然后在node2上执行sbin/start-master.sh启动第二个Master

Spark结构以及任务提交

Spark程序：Application
用于提交应用程序：driver 驱动类
Driver 作用创建RDD 构建DAG
资源管理者：Master
节点管理者：Worker
真正执行程序者：Executor (进程)
Spark 基础 - 图7

执行Spark程序

执行第一个spark程序

  1. /root/apps/spark-1.5.2-bin-hadoop2.6/bin/spark-submit \
  2. --class org.apache.spark.examples.SparkPi \
  3. --master spark://node1.sfy.cn:7077 \
  4. --executor-memory 1G \
  5. --total-executor-cores 2 \
  6. /root/apps/spark-1.6.1-bin-hadoop2.6/lib/spark-examples-1.6.1-hadoop2.6.0.jar \
  7. 100

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

启动Spark Shell

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

启动spark shell

  1. /root/apps/spark-1.6.1-bin-hadoop2.6/bin/spark-shell \
  2. --master spark://node1.sfy.cn:7077 \
  3. --executor-memory 2g \
  4. --total-executor-cores 2

Spark 基础 - 图8
参数说明：
—master spark://cdh1:7077 指定Master的地址
—executor-memory 2g 指定每个worker可用内存为2G
—total-executor-cores 2 指定整个集群使用的cup核数为2个
注意：
如果启动spark shell时没有指定master地址，但是也可以正常启动spark shell和执行spark shell中的程序，其实是启动了spark的local模式，该模式仅在本机启动一个进程，没有与集群建立联系。

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

在spark shell中编写WordCount程序

  1. 首先启动hdfs

  2. 向hdfs上传一个文件到hdfs://cdh1:9000/words.txt

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

sc.textFile(“hdfs://cdh1:9000/words.txt”).flatMap(.split(“ “))
.map((,1)).reduceByKey(+).saveAsTextFile(“hdfs://node1.sfy.cn:9000/out”)
Spark 基础 - 图9

  1. 使用hdfs命令查看结果

hdfs dfs -ls hdfs://cdh1:9000/out/p*
说明：
sc 是SparkContext对象，该对象时提交spark程序的入口
textFile(hdfs://cdh1:9000/words.txt)是hdfs中读取数据
flatMap(.split(“ “))先map在压平
map((,1))将单词和1构成元组
reduceByKey(+)按照key进行reduce，并将value累加
saveAsTextFile(“hdfs://node1.sfy.cn:9000/out”)将结果写入到hdfs中

在IDEA中编写WordCount程序

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

  1. package com.zhiyoulxj.spark
  3. import org.apache.spark.{SparkConf, SparkContext}
  5. object WordCount {
  6.   def main(args: Array[String]): Unit = {
  7.     //spark的入口
  8.     val cfg = new SparkConf().setAppName("WC").setMaster("local[2]")
  9.     val sc = new SparkContext(cfg)
  10.     sc.textFile("F:\\words.txt").flatMap(_.split("")).map((_,1))
  11.       .reduceByKey(_+_).sortBy(_._2,false).saveAsTextFile("F:\\words")
  12.     sc.stop()
  13.   }
  14. }

架包依赖

  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <project xmlns="http://maven.apache.org/POM/4.0.0"
  3.          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  4.          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  5.     <modelVersion>4.0.0</modelVersion>
  7.     <groupId>com.zhiyoulxj</groupId>
  8.     <artifactId>SparkWordCount</artifactId>
  9.     <version>1.0-SNAPSHOT</version>
  10.     <properties>
  11.         <maven.compiler.source>1.7</maven.compiler.source>
  12.         <maven.compiler.target>1.7</maven.compiler.target>
  13.         <encoding>UTF-8</encoding>
  14.         <scala.version>2.10.6</scala.version>
  15.         <scala.compat.version>2.10</scala.compat.version>
  16.     </properties>
  18.     <dependencies>
  19.         <dependency>
  20.             <groupId>org.scala-lang</groupId>
  21.             <artifactId>scala-library</artifactId>
  22.             <version>${scala.version}</version>
  23.         </dependency>
  25.         <dependency>
  26.             <groupId>org.apache.spark</groupId>
  27.             <artifactId>spark-core_2.10</artifactId>
  28.             <version>1.5.2</version>
  29.         </dependency>
  31.         <dependency>
  32.             <groupId>org.apache.spark</groupId>
  33.             <artifactId>spark-streaming_2.10</artifactId>
  34.             <version>1.5.2</version>
  35.         </dependency>
  37.         <dependency>
  38.             <groupId>org.apache.hadoop</groupId>
  39.             <artifactId>hadoop-client</artifactId>
  40.             <version>2.6.2</version>
  41.         </dependency>
  42.     </dependencies>
  44.     <build>
  45.         <sourceDirectory>src/main/scala</sourceDirectory>
  46.         <testSourceDirectory>src/test/scala</testSourceDirectory>
  47.         <plugins>
  48.             <plugin>
  49.                 <groupId>net.alchim31.maven</groupId>
  50.                 <artifactId>scala-maven-plugin</artifactId>
  51.                 <version>3.2.0</version>
  52.                 <executions>
  53.                     <execution>
  54.                         <goals>
  55.                             <goal>compile</goal>
  56.                             <goal>testCompile</goal>
  57.                         </goals>
  58.                         <configuration>
  59.                             <args>
  60.                                 <arg>-make:transitive</arg>
  61.                                 <arg>-dependencyfile</arg>
  62.                                 <arg>${project.build.directory}/.scala_dependencies</arg>
  63.                             </args>
  64.                         </configuration>
  65.                     </execution>
  66.                 </executions>
  67.             </plugin>
  68.             <plugin>
  69.                 <groupId>org.apache.maven.plugins</groupId>
  70.                 <artifactId>maven-surefire-plugin</artifactId>
  71.                 <version>2.18.1</version>
  72.                 <configuration>
  73.                     <useFile>false</useFile>
  74.                     <disableXmlReport>true</disableXmlReport>
  75.                     <includes>
  76.                         <include>**/*Test.*</include>
  77.                         <include>**/*Suite.*</include>
  78.                     </includes>
  79.                 </configuration>
  80.             </plugin>
  82.             <plugin>
  83.                 <groupId>org.apache.maven.plugins</groupId>
  84.                 <artifactId>maven-shade-plugin</artifactId>
  85.                 <version>2.3</version>
  86.                 <executions>
  87.                     <execution>
  88.                         <phase>package</phase>
  89.                         <goals>
  90.                             <goal>shade</goal>
  91.                         </goals>
  92.                         <configuration>
  93.                             <filters>
  94.                                 <filter>
  95.                                     <artifact>*:*</artifact>
  96.                                     <excludes>
  97.                                         <exclude>META-INF/*.SF</exclude>
  98.                                         <exclude>META-INF/*.DSA</exclude>
  99.                                         <exclude>META-INF/*.RSA</exclude>
  100.                                     </excludes>
  101.                                 </filter>
  102.                             </filters>
  103.                             <transformers>
  104.                                 <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
  105.                                     <mainClass>cn.sfy.spark.WordCount</mainClass>
  106.                                 </transformer>
  107.                             </transformers>
  108.                         </configuration>
  109.                     </execution>
  110.                 </executions>
  111.             </plugin>
  112.         </plugins>
  113.     </build>
  114. </project>

1.首先启动hdfs和Spark集群
启动hdfs
[root@cdh1 ~]# start-dfs.sh
启动spark
[root@cdh1 ~]# start-spark-all.sh

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

  1. /root/apps/spark-1.6.1-bin-hadoop2.6/bin/spark-submit \
  2. --class cn.sfy.spark.WordCount \
  3. --master spark://node1.sfy.cn:7077 \
  4. --executor-memory 2G \
  5. --total-executor-cores 1 \
  6. /root/spark-mvn-1.0-SNAPSHOT.jar \
  7. hdfs://cdh1:9000/words.txt \
  8. hdfs://cdh1:9000/out001

3.查看程序执行结果
hdfs dfs -cat hdfs://cdh1:9000/out/part-00000
(hello,6)
(tom,3)
(kitty,2)
(jerry,1)

RDD练习

启动spark-shell

  1. /root/apps/spark-1.6.1-bin-hadoop2.6/bin/spark-shell --master spark://cdh1:7077

练习1：

1、通过并行化生成rdd
val rdd1 = sc.parallelize(List(5, 6, 4, 7, 3, 8, 2, 9, 1, 10))
Spark 基础 - 图10
2、对rdd1里的每一个元素乘2然后排序
val rdd2 = rdd1.map( * 2).sortBy(x => x, true)
Spark 基础 - 图11
3、过滤出大于等于十的元素
val rdd3 = rdd2.filter( >= 10)
Spark 基础 - 图12
4、字典序排列
Spark 基础 - 图13

练习2：

val rdd1 = sc.parallelize(Array(“a b c”, “d e f”, “h i j”))
Spark 基础 - 图14
将rdd1里面的每一个元素先切分在压平
val rdd2 = rdd1.flatMap(_.split(‘ ‘))
rdd2.collect
Spark 基础 - 图15

练习3：

val rdd1 = sc.parallelize(List(5, 6, 4, 3))
val rdd2 = sc.parallelize(List(1, 2, 3, 4))
Spark 基础 - 图16
1、求并集
val rdd3 = rdd1.union(rdd2)
Spark 基础 - 图17
2、求交集
val rdd4 = rdd1.intersection(rdd2)
Spark 基础 - 图18
3、去重
rdd3.distinct.collect
Spark 基础 - 图19

练习4：

val rdd1 = sc.parallelize(List((“tom”, 1), (“jerry”, 3), (“kitty”, 2)))
val rdd2 = sc.parallelize(List((“jerry”, 2), (“tom”, 1), (“shuke”, 2)))
Spark 基础 - 图20
1、求jion
val rdd3 = rdd1.join(rdd2)
rdd3.collect
Spark 基础 - 图21
2、求并集
val rdd4 = rdd1 union rdd2
Spark 基础 - 图22
3、按key进行分组
rdd4.groupByKey.collect
Spark 基础 - 图23
4、左外连接
Spark 基础 - 图24

练习5：

val rdd1 = sc.parallelize(List((“tom”, 1), (“tom”, 2), (“jerry”, 3), (“kitty”, 2)))
val rdd2 = sc.parallelize(List((“jerry”, 2), (“tom”, 1), (“shuke”, 2)))
Spark 基础 - 图25
//cogroup
val rdd3 = rdd1.cogroup(rdd2).collect
//注意cogroup与groupByKey的区别
Spark 基础 - 图26

练习6：

val rdd1 = sc.parallelize(List(1, 2, 3, 4, 5))
Spark 基础 - 图27
//reduce聚合
val rdd2 = rdd1.reduce( + )
Spark 基础 - 图28

练习7：

val rdd1 = sc.parallelize(List((“tom”, 1), (“jerry”, 3), (“kitty”, 2), (“shuke”, 1)))
val rdd2 = sc.parallelize(List((“jerry”, 2), (“tom”, 3), (“shuke”, 2), (“kitty”, 5)))
Spark 基础 - 图29
1、并集
val rdd3 = rdd1.union(rdd2)
Spark 基础 - 图30
rdd3.collect
Spark 基础 - 图31
2、按key进行聚合
val rdd4 = rdd3.reduceByKey( + )
Spark 基础 - 图32
Spark 基础 - 图33
3、排序
val rdd5 = rdd4.map(t => (t._2, t._1)).sortByKey(false).map(t => (t._2, t._1))
Spark 基础 - 图34
rdd5.collect
Spark 基础 - 图35

求和1

1、val rdd1 = sc.parallelize(List((“tom”, 1), (“jerry”, 3), (“kitty”, 2)))
val rdd2 = sc.parallelize(List((“jerry”, 2), (“tom”, 1), (“shuke”, 2)))
Spark 基础 - 图36
2、val rdd3 = rdd1.union(rdd2)
Spark 基础 - 图37
3、rdd3.groupByKey.map(x=>(x._1,x._2.sum))
rdd3.groupByKey.map(x=>(x._1,x._2.sum)).collect
Spark 基础 - 图38

求和2 从文件中读取

倒叙false sortBy(_._2,false)

reduceByKey() 速度较快(相对于groupByKey()) 会在本地先进行一个局部的汇总

  1. sc.textFile("/root/words.txt").flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).sortBy(_._2,false).collect

Spark 基础 - 图39

groupByKey()

  1. sc.textFile("/root/words.txt").flatMap(_.split(" ")).map((_,1)).groupByKey.map(t=>(t._1,t._2.sum)).sortBy(_._2,false).collect

Spark 基础 - 图40

求和3

1、val rdd1 = sc.parallelize(List((“tom”, 1), (“tom”, 3), (“jack”, 2),(“rose”,1)))
Spark 基础 - 图41
val rdd2 = sc.parallelize(List((“jack”, 1), (“tom”, 2), (“jerry”, 2),(“kitty”,1)))
Spark 基础 - 图42
cogroup相当于SQL中的全外关联full outer join，返回左右RDD中的记录，关联不上的为空。
2、rdd1.cogroup(rdd2).collect
Spark 基础 - 图43
3、rdd1.cogroup(rdd2).map(t=>(t._1,t._2._1.sum+t._2._2.sum)).collect
Spark 基础 - 图44

笛卡尔积

1、val rdd1 = sc.parallelize(List((“tom”, 1), (“jack”, 1)))
Spark 基础 - 图45
2、val rdd2 = sc.parallelize(List((“tom”, 1), (“jerry”, 2)))
Spark 基础 - 图46
3、rdd1.cartesian(rdd2).collect
Spark 基础 - 图47

集合元素操作

1、集合中元素个数
Spark 基础 - 图48
2、取出列表中值最大的两个
Spark 基础 - 图49
3、在列表中从头开始按顺序取出元素
Spark 基础 - 图50
4、取出第一个元素
Spark 基础 - 图51
5、在升序序列中从头开始取出元素 takerOrdered（3）有序取出三个元素
Spark 基础 - 图52