首先是一张Spark的部署图:
节点类型有:
1. master 节点: 常驻master进程,负责管理全部worker节点。
2. worker 节点: 常驻worker进程,负责管理executor 并与master节点通信。
dirvier:官方解释为: The process running the main() function of the application and creating the SparkContext。即理解为用户自己编写的应用程序
一、Application
application(应用)其实就是用spark-submit提交的程序。比方说spark examples中的计算pi的SparkPi。一个application通常包含三
部分:从数据源(比方说HDFS)取数据形成RDD,通过RDD的transformation和action进行计算,将结果输出到console或者外部存
储(比方说collect收集输出到console)。
二、Driver
主要完成任务的调度以及和executor和cluster manager进行协调。有client和cluster联众模式。client模式driver在任务提交的机器上
运行,而cluster模式会随机选择机器中的一台机器启动driver。从spark官网截图的一张图可以大致了解driver的功能。
三、Executor
在每个WorkerNode上为某应用启动的一个进程,该进程负责运行任务,并且负责将数据存在内存或者磁盘上,每个任务都有各自独
立的Executor。
Executor是一个执行Task的容器。它的主要职责是:
1、初始化程序要执行的上下文SparkEnv,解决应用程序需要运行时的jar包的依赖,加载类。
2、同时还有一个ExecutorBackend向cluster manager汇报当前的任务状态,这一方面有点类似Hadoop的tasktracker和task。
总结:Executor是一个应用程序运行的监控和执行容器。Executor的数目可以在submit时,由 --num-executors (on yarn)指定.
四、Job
包含很多task的并行计算,可以认为是Spark RDD 里面的action,每个action的计算会生成一个job。
用户提交的Job会提交给DAGScheduler,Job会被分解成Stage和Task。
五、Task
即 stage 下的一个任务执行单元,一般来说,一个 rdd 有多少个 partition,就会有多少个 task,因为每一个 task 只是处理一个
partition 上的数据。
每个executor执行的task的数目, 可以由submit时,--num-executors(on yarn) 来指定。
六、Stage
一个Job会被拆分为多组Task,每组任务被称为一个Stage就像Map Stage, Reduce Stage。
Stage的划分在RDD的论文中有详细的介绍,简单的说是以shuffle和result这两种类型来划分。在Spark中有两类task,一类是
shuffleMapTask,一类是resultTask,第一类task的输出是shuffle所需数据,第二类task的输出是result,stage的划分也以此为依
据,shuffle之前的所有变换是一个stage,shuffle之后的操作是另一个stage。比如 rdd.parallize(1 to 10).foreach(println) 这个操作没
有shuffle,直接就输出了,那么只有它的task是resultTask,stage也只有一个;如果是rdd.map(x => (x, 1)).reduceByKey(_ +
_).foreach(println), 这个job因为有reduce,所以有一个shuffle过程,那么reduceByKey之前的是一个stage,执行shuffleMapTask,
输出shuffle所需的数据,reduceByKey到最后是一个stage,直接就输出结果了。如果job中有多次shuffle,那么每个shuffle之前都是
一个stage。
换句话来讲,就要提到spark里面的宽依赖和窄依赖:
看一下父RDD中的数据是否进入不同的子RDD,如果只进入到一个子RDD则是窄依赖,否则就是宽依赖。宽依赖和窄依赖的边界就
是stage的划分点。从spark的论文中的两张截图,可以清楚的理解宽窄依赖以及stage的划分。
上面大图大字看得有点晕,下面这幅图是找到的资料中比较清晰的:
Application:Spark 的应用程序,用户提交后,Spark为App分配资源,将程序转换并执行,其中Application包含一个Driver program和若干Executor
SparkContext:Spark 应用程序的入口,负责调度各个运算资源,协调各个 Worker Node 上的 Executor
Driver Program:运行Application的main()函数并且创建SparkContext
RDD Graph:RDD是Spark的核心结构, 可以通过一系列算子进行操作(主要有Transformation和Action操作)。当RDD遇到Action算子时,将之前的所有算子形成一个有向无环图(DAG)。再在Spark中转化为Job,提交到集群执行。一个App中可以包含多个Job
Executor:是为Application运行在Worker node上的一个进程,该进程负责运行Task,并且负责将数据存在内存或者磁盘上。每个Application都会申请各自的Executor来处理任务
Worker Node:集群中任何可以运行Application代码的节点,运行一个或多个Executor进程
Job:一个RDD Graph触发的作业,往往由Spark Action算子触发,在SparkContext中通过runJob方法向Spark提交Job
Stage:每个Job会根据RDD的宽依赖关系被切分很多Stage, 每个Stage中包含一组相同的Task, 这一组Task也叫TaskSet
Task:一个分区对应一个Task,Task执行RDD中对应Stage中包含的算子。Task被封装好后放入Executor的线程池中执行
DAGScheduler:根据Job构建基于Stage的DAG,并提交Stage给TaskScheduler
TaskScheduler:将Taskset提交给Worker node集群运行并返回结果
从经验角度来看,关于spark作业配置初始化参数应该参照:
在集群中,建议为每一个 CPU 核( core )分配 3-4 个任务。
由于spark streaming是基于simple consumer api消费kafka topic,所以一个topic有多少个partition,就有多少个task。假设我们消费
的这个topic总共有40个partition,所以起初spark streaming的task数量为40。根据上面的建议 一个CPU核分3到4个任务(这个值具体看topic消息量,
如果有很多消息的话可以适当降低一个cpu的任务量),则初始化
时我们需要10个core。一个executor可以分配2到5个core,executor的memory根据处理的数据量自行设置,一般推荐2G
如果使用了spark sql或者很多数据处理是在Driver端,可以适当的调大Driver的memory,也推荐2G,Cpu默认使用一个core
所以上面的例子最后的配置为:driver-memory=2G,driver-cores=1,num-executors=2或者5,executor-cores=5或者2,executor-
memory=2G
【转载自:http://blog.csdn.net/WangQYoho/article/details/70210486】