一、Spark-Core（二）回顾

二、Stage剖析

• 2.1、遇到action产生job
• 2.2、job产生stage
• 2.3、rdd中的cache
• 2.4、Spark-shell中测试rdd缓存 && StorageLevel
• 2.5、Spark-Core中的框架选择（MEMORY_ONLY）
• 2.6、recomputing重算概念
• 2.7、Spark中的宽窄依赖

三、Spark的运行架构（重要指数五颗星）

• 3.1 总结
• 3.2 Cluster Mode Overview

一、Spark-Core（二）回顾

• Spark的运行架构，各种关键术语的解释：Driver、Cluster Manager、Executor

RDD中有很多方法，a list of partitions，有一个函数：a function for computing each split，有一系列的依赖： a list of dependencies on other RDD；

五大特性	对应的在源码中的方法	运行在driver端还是executor端	Input	Output
A list of partitions	getPartitions	？	Partition
A function for compting each split	compute	？	Iterable可迭代的
A list of dependencies on other RDD	getDependencies	？	Dependency

可选的操作：a Partitioner for key-value RDDs、A list of preferred locations to compute each split

二、Stage剖析：

2.1、遇到action产生job

1、每遇到一个action就会触发一个job，每一个job又会被拆分成更小的task

在Spark-shell中执行如下：

scala> sc.parallelize(List(1,2,3,4,4,4,55,55,2,1)).map((_,1)).reduceByKey(_+_).collect
res1: Array[(Int, Int)] = Array((4,3), (2,2), (55,2), (1,2), (3,1))             scala> sc.parallelize(List(1,2,3,4,4,4,55,55,2,1)).map((_,1)).collect
res2: Array[(Int, Int)] = Array((1,1), (2,1), (3,1), (4,1), (4,1), (4,1), (55,1), (55,1), (2,1), (1,1))

• WebUI界面的结果：
  
  Job的定义：
  1、 一种由多个task组成的并行计算，job由Spark中相应算子进行触发（例如save、collect）；你会在driver’s log中看到这个术语：
• A parallel computation consisting of multiple tasks that gets spawned in response to a Spark action (e.g. save, collect); you’ll see this term used in the driver’s logs

1. 每遇到一个action算子如collect就会变成一个job
2. 如上图所示：我们两个计算中使用了两个collect算子，所以生成了两个job
3. Job id的初始值是0，为什么接下来是1，然后是2；因为每遇到一个action后job id的值都会依次递增1；

2.2、Job如何产生Stage

1、每一个job都会被拆分成更小的task ==>称为Stages；如果Stage之间有依赖，必须前一个Stage执行完后才执行后一个Stage；

• Each job get divided into smaller sets of tasks called stages that depend on each other(similar to the map and reduce stages in MapReduce)

1、下图中的DAG图就是这句语句中出现的：
sc.parallelize(List(1,1,2,2,3,3,3,3,4,4,4,5)).map((,1)).reduceByKey(+_).collect
res2: Array[(Int, Int)] = Array((4,3), (2,2), (1,2), (3,4), (5,1))2、reduceByKey算子会产生shuffle，一遇到shuffle就会产生Stage；stage2需要等stage1执行完才执行3、如上语句中分别是如下的3个算子：parallelize、map、reduceByKey；reduceByKey算子会产生shuffle，shuffle会产生stage。举例：原来是一个stage，当我们遇到shuffle后，就会被切一刀，变成2个stage。

遇到collect触发成为1个job，然后job中有带shuffle的reduceByKey又被拆分为2个stages；如下图中的collect at和map at都是以stage中的最后一个算子进行命名的。

2.3、Rdd中的Cache

Rdd中的缓存主要是用于提升速度使用，扩充：JVM：java memory model，计算是通过cpu来处理的，数据是存在内存中的，现在很多地方都很耗费cpu；

Spark中最重要的一个功能是持久化数据到内存中，内存存储在executor中

1、Spark非常重要的一个功能是将rdd持久化在内存中。当对rdd执行持久化操作的时候，每个节点都会将自己操作的RDD的partition持久化到内存中；这样的话，针对于一个rdd反复操作的场景，就只要对rdd进行一次计算即可，后面再使用该rdd不需要反复计算了。

2、 要持久化RDD，只要调用其cache()或者persist()方法即可。在该rdd第一次被计算出来的时候，就会直接缓存在每个节点中。而且持久化机制是容错的，如果持久化的rdd中的任何partition丢失了，那么spark还会通过其源rdd，使用transformation操作重新计算该partition。

3、 cache和persist的区别在于，cache调用的就是persist，而persist调用的是persist(memory_only)；如果需要在内存中清除缓存，采用unpersist方法。

2.4、Spark-shell中测试rdd缓存 && StorageLevel

1、读取文件：
scala> val lines = sc.textFile("file:///home/hadoop/data/ruozeinput.txt")
lines: org.apache.spark.rdd.RDD[String] = file:///home/hadoop/data/ruozeinput.txt MapPartitionsRDD[1] at textFile at <console>:242、把这个文件cache住：
scala> lines.cache
res0: lines.type = file:///home/hadoop/data/ruozeinput.txt MapPartitionsRDD[1] at textFile at <console>:243、使用collect进行触发：
scala> lines.collect
res1: Array[String] = Array(hello       hello   hello, world    world, john)

1、使用cache的时候不触发job，cache操作做是lazy的，它遇到collect算子后才会触发；在storage页面中查看到的是缓存信息，有2个分区被100% cache住了；

2、在Stage页面中查看到输入文件大小是53B，而在内存中该文件大小是240B，反而变大了

可以联想到Spark-Core2中的LogAPP解析需求：
一个作业中多个需求，有些部分可以抽取出来，把它cache住，避免反复调用；

cache和persist的方法的区别：

1、cache调用的是persist方法

2、persist调用的是persist中的一个只读内存方法

SparkCore默认使用的就是memory_only方法：

两个方法在源码中的区别：

* Persist this RDD with the default storage level (`MEMORY_ONLY`).def persist(): this.type = persist(StorageLevel.MEMORY_ONLY)* Persist this RDD with the default storage level (`MEMORY_ONLY`).def cache(): this.type = persist()//鼠标ctrl+左键，点进StorageLevel中去：
object StorageLevel {val NONE = new StorageLevel(false, false, false, false)val DISK_ONLY = new StorageLevel(true, false, false, false)val DISK_ONLY_2 = new StorageLevel(true, false, false, false, 2)val MEMORY_ONLY = new StorageLevel(false, true, false, true)val MEMORY_ONLY_2 = new StorageLevel(false, true, false, true, 2)val MEMORY_ONLY_SER = new StorageLevel(false, true, false, false)val MEMORY_ONLY_SER_2 = new StorageLevel(false, true, false, false, 2)val MEMORY_AND_DISK = new StorageLevel(true, true, false, true)val MEMORY_AND_DISK_2 = new StorageLevel(true, true, false, true, 2)val MEMORY_AND_DISK_SER = new StorageLevel(true, true, false, false)val MEMORY_AND_DISK_SER_2 = new StorageLevel(true, true, false, false, 2)val OFF_HEAP = new StorageLevel(true, true, true, false, 1)private var _useDisk: Boolean,		磁盘private var _useMemory: Boolean,		内存private var _useOffHeap: Boolean,		堆外内存private var _deserialized: Boolean,		反序列化	private var _replication: Int = 1)		副本数

清除缓存：

• scala> lines.unpersist()
  res2: lines.type = file:///home/hadoop/data/ruozeinput.txt MapPartitionsRDD[1] at textFile at :24

执行cache的时候是一个lazy操作的，清除缓存的时候却是eager的；

Spark-shell中测试StorageLevel

scala> import org.apache.spark.storage.StorageLevel
import org.apache.spark.storage.StorageLevelscala> lines.persist(StorageLevel.MEMORY_AND_DISK_SER_2)
res3: lines.type = file:///home/hadoop/data/ruozeinput.txt MapPartitionsRDD[1] at textFile at <console>:24scala> lines.count
20/06/11 13:51:12 WARN RandomBlockReplicationPolicy: Expecting 1 replicas with only 0 peer/s.
20/06/11 13:51:12 WARN RandomBlockReplicationPolicy: Expecting 1 replicas with only 0 peer/s.
20/06/11 13:51:12 WARN BlockManager: Block rdd_1_0 replicated to only 0 peer(s) instead of 1 peers
20/06/11 13:51:12 WARN BlockManager: Block rdd_1_1 replicated to only 0 peer(s) instead of 1 peers
res4: Long = 3

2.5、Spark-Core中的框架选择

1、http://spark.apache.org/docs/latest/rdd-programming-guide.html，找到rdd-persisitance

Storage Level	Meaning
MEMORY_ONLY	Store RDD as deserialized Java objects in the JVM. If the RDD does not fit in memory, some partitions will not be cached and will be recomputed on the fly each time they’re needed. This is the default level.
MEMORY_ONLY_SER
(Java and Scala)	Store RDD as serialized Java objects (one byte array per partition). This is generally more space-efficient than deserialized objects, especially when using a fast serializer, but more CPU-intensive to read.

此处需要了解到的：
1、缓存可以使用
2、内存和内存序列号

2.6、recomputing重算的概念

生产场景举例：
1、假设一个rdd，数据在hdfs上，使用textFile把数据读取进来，做了map操作后变成了一个新的rdd，而后触发了一个action；
-我们知道的对一个rdd做计算就是对rdd中的每一个分区做计算，在rdd1中做了一个reduceByKey或groupByKey的操作，就变成了两个rdd；分区中的数据存放位置不确定；

2、假设rdd1中的第三个partition挂掉了，根据我们rdd的依赖关系，spark会找到父rdd的信息，这个叫做血缘关系。

2.7、Spark中的宽窄依赖

对上图需要进行拆分，reduceByKey前的是stage0，reduceByKey后的是stage1；这就涉及到了依赖，Dependency依赖：

Spark中有两大依赖：宽窄依赖

Narrow：窄依赖
定义：一个父rdd的partition只能被子RDD的某个partition使用一次（map filter union join with input co-partitioned）

Wide：宽依赖，带shuffle的
定义：一个父rdd的partition能被子rdd的partition使用多次

groupByKey：一份数据会被子rdd的partition使用多次

对于宽窄依赖的区别：

• 对于窄依赖：父rdd的某个partition丢了，问题不大，只需要把父rdd的某个partition单独算出来就行了；
• 对于宽依赖，父rdd的partition丢了，所有的都需要重新计算。

第二张图分析：

解析：
A遇到groupBy拆分一个stage，黑色区域理解为挂了，join的时候要根据key来shuffle的，在action之前遇到shuffle算子就会变成2个stage；

action产生job，job由N个stage构成，stage由N个task构成；

MapReduce：1+1+1+1
1+1 ==》 2
2+1 ==》 3
3+1 ==》4
对于窄依赖来说，以pipeline的方式一条路干到底；

2.8、Working with key-value Pairs

1、通过key-value访问，最常用的方式是分布式中的shuffle操作，就像group通过key聚合的操作

2、The Key-value operations are available in the PairRDD class, which automatically wraps around an RDD of tuples.

我们在写wordcount的时候并没有用到pairRDD，文本中是单词，为每个单词赋上1个1，然后一个reduceByKey进行分发；

查看PairRDDFunctions.scala的源码：

1、reduceByKey就是PairRDDFunctions这个类中的：/*** Merge the values for each key using an associative and commutative reduce function. This will* also perform the merging locally on each mapper before sending results to a reducer, similarly* to a "combiner" in MapReduce. Output will be hash-partitioned with the existing partitioner/* parallelism level.*/def reduceByKey(func: (V, V) => V): RDD[(K, V)] = self.withScope {reduceByKey(defaultPartitioner(self), func)}2、在RDD.scala中查看这个方法：implicit def rddToPairRDDFunctions[K, V](rdd: RDD[(K, V)])(implicit kt: ClassTag[K], vt: ClassTag[V], ord: Ordering[K] = null): PairRDDFunctions[K, V] = {new PairRDDFunctions(rdd)}//人到超人传一个人进来，里面其实是new了一个超人出去，把人放到里面

测试程序中有reduceByKey算子会不会使用RDD.scala中的rddToPairRDDFunctions这个方法：
在这个地方打个断点，debug测试运行：

面试题：
reduceByKey是哪个类中的算子？

• PairRDDFunctions这个类