一、Spark作业调度方式
1、local 测试或实验性质的本地运行模式
local[N] 是用单机的多个线程来模拟Spark分布式计算，通常用来验证开发出来的应用程序逻辑上有没有问题。
其中N代表可以使用N个线程，每个线程拥有一个core。如果不指定N，则默认是1个线程（该线程有1个core）。
example:
spark-submit
--class JavaWordCount
--master local[10]
JavaWordCount.jar
file:///tmp/test.txt
2、standalone：指定节点
使用SparkSubmit提交任务的时候(包括Eclipse或者其它开发工具使用new SparkConf()来运行任务的时候)，Driver运行在Client
spark-submit
--master spark://wl1:6066
--deploy-mode cluster
3、Yarn
1)yarn client 测试中常用
Spark跑在Hadoop集群中，所以为了做到资源能够均衡调度，会使用YARN来做为Spark的Cluster Manager，来为Spark的应用程序分配资源。
在执行Spark应用程序前，要启动Hadoop的各种服务。由于已经有了资源管理器，所以不需要启动Spark的Master、Worker守护进程
Driver是在客户端运行
spark-submit --master yarn --deploy-mode client
2）yarn cluster 生产中用
spark-submit脚本提交，向yarn（RM）中提交ApplicationMaster程序、AM启动的命令和需要在Executor中运行的程序等
Resource Manager在集群中的某个NodeManager上运行ApplicationMaster，该AM同时会执行driver程序。紧接着，会在各NodeManager上运行
spark-submit --master yarn --deploy-mode cluster
cluster和client的区别：
cluster:sparkcontext创建在yarn集群中
client:sparkcontext创建在本地
二、SparkRDD
1、Spark作业按照算子表示执行过程
2、 分类
1、transformation
概念：字面意思就是进行转换，将rdd由一个形态转化成另一个形态
常见的tranformation算子
flatMap:将行拆分为单词
map：是最常用的算子，将将原rdd的形态，转化为另外一种形态。
filter：过滤
sample:根据给定的随机种子seed,随机抽样出数量为frac的数据
union：返回一个新的数据集，由原数据集和参数联合而成 该union操作和sql中的union all操作一模一样（不去重）
groupByKey：对数组进行 group by key操作 慎用
reduceByKey:需要先进行一次本地的预聚合，每一个mapper(对应的partition)执行一次预聚合
join:打印关联的组合信息
sortByKey：进行排序
combineBykey：
注意：
使用combineByKey和aggregateByKey来模拟groupByKey和reduceByKey
不管combineByKey还是aggregateByKey底层都是使用combineByKeyWithClassTag来实现的
区别：
1、本质上combineByKey和aggregateByKey都是通过combineByKeyWithClassTag来实现的，只不过实现的细节或者方式不大一样。
2、combineByKey更适合做聚合前后数据类型不一样的操作，aggregateByKey更适合做聚合前后数据类型一致的操作
因为我们可以在combineByKey提供的第一个函数中完成比较复杂的初始化操作，而aggregateByKey的第一个参数是一个值
3、我们使用时最简单的版本，而在实际生产过程中，一般都是相对比较复杂的版本，还有其它参数的，比如partitioner
2、action算子
概念：字面意思就是执行，action的操作目的就是用来执行一个job作业的
常见的action算子
reduce：执行reduce操作，返回值是一个标量
collect :将数据从集群中的worker上拉取到dirver中，所以在使用的过程中要慎用，意外拉取数据过大造成driver内存溢出
count :返回当前rdd中有多少条记录
take : 获取rdd前n条记录
first : take(1) 获取rdd中的第一条记录
savenAsTextFile:将RDD中的数据保存到文件系统中
countByKey:和readuceByKey效果相同，但reduceByKey是一个Transformation
foreach:遍历
3、RDD缓存
Spark 速度非常快的原因之一，就是在不同操作中可以在内存中持久化或缓存个数据集。当持久化某个 RDD 后，每一个节点都将把计算的分片结果保存在内存中，并在对此 RDD 或衍生出的 RDD 进行的其他动作中重用。这使得后续的动作变得更加迅速。RDD 相关的持久化和缓存，是 Spark 最重要的特征之一。可以说，缓存是 Spark 构建迭代式算法和快速交互式查询的关键。
RDD 的缓存方式
RDD 通过 persist 方法或 cache 方法可以将前面的计算结果缓存，但是并不是这两个方法被调用时立即缓存，而是触发后面的 action 时，该 RDD 将会被缓存在计算节点的内存中，并供后面重用。
通过查看源码发现 cache 最终也是调用了 persist 方法，默认的存储级别都是仅在内存存储一份，Spark 的存储级别还有好多种，存储级别在 object StorageLevel 中定义的
4、Shared Variables（共享变量）
在 Spark 程序中，当一个传递给 Spark 操作(例如 map 和 reduce)的函数在远程节点上面运行时，Spark 操作实际上操作的是这个函数所用变量的一个独立副本。这些变量会被复制到每 台机器上，并且这些变量在远程机器上的所有更新都不会传递回驱动程序。通常跨任务的读写变量是低效的，但是，Spark 还是为两种常见的使用模式提供了两种有限的共享变量：广播变（Broadcast Variable）和累加器（Accumulator）
1）Broadcast Variables（广播变量）
如果我们要在分布式计算里面分发大对象，例如：字典，集合，黑白名单等，这个都会由Driver 端进行分发，一般来讲，如果这个变量不是广播变量，那么每个 task 就会分发一份，这在 task 数目十分多的情况下 Driver 的带宽会成为系统的瓶颈，而且会大量消耗 task 服务器上的资源，如果将这个变量声明为广播变量，那么知识每个 executor 拥有一份，这个executor 启动的 task 会共享这个变量，节省了通信的成本和服务器的资源。
没有使用广播变量：
使用了广播变量之后：
如何定义和还原一个广播变量
定义 ： val broadcast = sc.broadcast(a)
还原： val c = broadcast.value
2）Accumulators（累加器）
在 Spark 应用程序中，我们经常会有这样的需求，如异常监控，调试，记录符合某特性的数据的数目，这种需求都需要用到计数器，如果一个变量不被声明为一个累加器，那么它将在被改变时不会在 driver 端进行全局汇总，即在分布式运行时每个 task 运行的只是原始变量的一个副本，并不能改变原始变量的值，但是当这个变量被声明为累加器后，该变量就会有分布式计数的功能。
正确的图解：
如果定义和还原一个累加器
定义 ： val a = sc.longAccumulator(0)
还原 ： val b = a.value
三、Spark作业执行过程
1、使用wordCount案例来说明spark作业的执行过程
对图的几点说明：
1）通过wordcount的执行流程，可以看到在执行reducekbykey的时候，在上游的rdd结束之后执行了一次类似于mr编程中的本地聚合combiner,这样可以减少网络中传递的数据量，提高spark作业的执行效率
2）从上图中看到，上下游rdd中对应的partition有依赖关系，spark中把rdd之间的依赖关系形成的这个图称之为rdd的lineage(血统)
启动的依赖关系，有两种：宽依赖和窄依赖
窄依赖：下游rdd中的partition只依赖直接上游rdd中的某一个人partition或常数个partition,把这种依赖关系称之为窄依赖
窄依赖对应的算子：窄依赖算子，比如常见的flatmap,map,filter,union
宽依赖：下游rdd中一个partition中的数据依赖上游rdd中所有的partition中的数据，把这种关系称之为宽依赖
宽依赖对应的算子：款依赖算子，比如常见：join,reduceByKey,sortByKey,groupByKey
注意：宽依赖算子或者窄依赖算子是针对transformation而言的，而action是触发transformation执行的动因
3）宽依赖非常重要，是spark作业阶段划分的标志代码从后往前推，从action开始，到第一个宽依赖算子结束，我们叫做finalstage,resultstage,在该宽依赖左右就化划分成了两个阶段，右侧就是我们刚才提到的resultstage，左侧再来一遍，一直往前找，找到一个宽依赖就形成一个阶段。在执行过程中，自然就有例外，如果从action往前推，到头都没有宽依赖，自动产生一个阶段，这个阶段就是ResultStage,在阶段表示的时候，从前往后推，第一个阶段stage0,以此类推
2、Spark作业执行过程（详细）
(1)、将我们编写的程序打成 jar 包
(2)、调用 spark-submit 脚本提交任务到集群上运行
(3)、运行 sparkSubmit 的 main 方法，在这个方法中通过反射的方式创建我们编写的主类的实例对象，然后调用 main 方法，开始执行我们的代码（注意，我们的 spark 程序中的 driver就运行在 sparkSubmit 进程中）
(4)、当代码运行到创建 SparkContext 对象时，那就开始初始化 SparkContext 对象了
(5)、在初始化 SparkContext 对象的时候，会创建两个特别重要的对象，分别是：DAGScheduler和 TaskScheduler
【DAGScheduler 的作用】将 RDD 的依赖切分成一个一个的 stage，然后将 stage 作为 taskSet
提交给 TaskScheduler
(6)、在构建 TaskScheduler 的同时，会创建两个非常重要的对象，分别是 DriverActor 和ClientActor
【clientActor 的作用】向 master 注册用户提交的任务
【DriverActor 的作用】接受 executor 的反向注册，将任务提交给 executor
(7)、当 ClientActor 启动后，会将用户提交的任务和相关的参数封装到 ApplicationDescription对象中，然后提交给 master 进行任务的注册
(8)、当 master 接受到 clientActor 提交的任务请求时，会将请求参数进行解析，并封装成Application，然后将其持久化，然后将其加入到任务队列 waitingApps 中（FIFO先进先出）
(9)、当轮到我们提交的任务运行时，就开始调用 schedule()，进行任务资源的调度
(10)、master 将调度好的资源封装到 launchExecutor 中发送给指定的 worker
(11)、worker 接受到 Master 发送来的 launchExecutor 时，会将其解压并封装到 ExecutorRunner中，然后调用这个对象的 start(), 启动 Executor
(12)、Executor 启动后会向 DriverActor 进行反向注册
(13)、driverActor 会发送注册成功的消息给 Executor
(14)、Executor 接受到 DriverActor 注册成功的消息后会创建一个线程池，用于执行 DriverActor发送过来的 task 任务
(15)、当属于这个任务的所有的 Executor 启动并反向注册成功后，就意味着运行这个任务的环境已经准备好了，driver 会结束 SparkContext 对象的初始化，也就意味着 new SparkContext 这句代码运行完成
(16)、当初始化 sc 成功后，driver 端就会继续运行我们编写的代码，然后开始创建初始的 RDD，然后进行一系列转换操作，当遇到一个 action 算子时，也就意味着触发了一个 job
(17)、driver 会将这个 job 提交给 DAGScheduler
(18)、DAGScheduler 将接受到的 job，从最后一个算子向前推导，将 DAG 依据宽依赖划分成一个一个的 stage，然后将 stage 封装成 taskSet，并将 taskSet 中的 task 提交给 DriverActor
(19)、DriverActor 接受到 DAGScheduler 发送过来的 task，会拿到一个序列化器，对 task 进行序列化，然后将序列化好的 task 封装到 launchTask 中，然后将 launchTask 发送给指定的Executor
(20)、Executor 接受到了 DriverActor 发送过来的 launchTask 时，会拿到一个反序列化器，对launchTask 进行反序列化，封装到 TaskRunner 中，然后从 Executor 这个线程池中获取一个线程，将反序列化好的任务中的算子作用在 RDD 对应的分区上
---------------------
作者：爱学习的小明-1993
原文：https://blog.csdn.net/weixin_43909426/article/details/86140586
网友评论