大家好,又见面了,我是你们的朋友全栈君。
1·,运行原理
看个流程图:
解说就是:
1,构建spark Application运行环境
2,sparkcontext向资源管理器注册
3,sparkContext向资源管理器申请运行Executor(执行器)
4,资源管理器分配executor
5,资源管理器启动executor
6,executor发送心跳至资源管理器
7,sparkContext构建DAG图
8,将DAG图分解成stage(taskSet)
9,把stage(taskSet)发送给TaskScheduler
10,Executor向sparkContext申请task
11,taskScheduler将task发送给Executor运行
12,同时,sparkContext 将应用程序代码发送给Executor
13,task在executor上运行,运行完毕释放所有资源。
二,
在上图中,发生箭头交叉就形成一个stage,其中与伴随这shuffle操作,并且这些算子(groupby,join)都是Action中的算子,map,union则属于Transformation中的算子。
理解算子的含义:
- hadoop只有map、Reduce这两个算子
- Spark提供了很多算子:
如上图,一个Job被拆分成若干个stage,每个stage执行一些计算,产生一些中间结果,他们的目的是最终生成这个JOB的计算结果,每个stage是一个taskset,包含若干个task。Task是spark最小的工作单元。在一个executor(执行器)上完成一个特定的事情。
三、窄依赖与宽依赖的判断方式,这里提供三种:
- 窄依赖:一个父RDD的一个partition最多被一个子RDD的一个partition使用。
- 宽依赖:一个父RDD的一个partition被多个子RDD的partition使用。
- 是否会发生shuffle操作,宽依赖会发生shuffle操作。也就是总结1,2,一个partition结果只被一个子paritition使用也就相当于没有发生shuffle操作。
- 也可以看有没有发生combine操作,不同的parititons被多个子RDD使用,必然发生合并操作。
四、理解下RDD是什么:全名“弹性分布式数据集”
- 可类比理解就是,hdfs上文件分片后的状态,比如使用splitline(),按行分割,则一行就是一个RDD。
- RDD是不可改变的分布式集合对象,因为它是加载的文件,明显我们不能对hdfs上的文件做增删改。
- 如 val lines=sc.textFile(“/home/aa.txt”);
这里lines即RDDs;
如果aa.txt很大,那么按照hdfs的文件写入,我们知道aa.txt是被按照64m放到不同的datanode节点上,在执行算子时在各个节点上分别处理各自的数据,可是我们操作的的对象都是lines这个变量,因此lines也即是这些节点数据的集合,即RDDS.
4,RDDs创建的二种方式:1,val rdds=sc.textFile();2,并行化处理,创建一个类似Array的容器,val Rdds=sc.parallelize(Array(1,2,3,4),4) (注,第二个参数4是parititons的个数
5,RDD.persist():持久化
默认每次在RDDs上进行action操作,spark都重新计算RDDs,如果想重复利用一个RDDs,可以使用RDD.persisit()。例如:对于同一个lines,我要进行一系列的转换,然后用count计算,如果我还想接着计算reduce,那么持久化就会利用前面的count的缓存数据计算reduce.最后,可以使用unpersist()方法从缓存中移除。
6,RDDs的血统关系图:spark维护者RDDS之间的依赖关系的创建关系,叫做血统关系图。Spark使用血统关系图来计算每个RDD的需求和恢复丢失的数据。
上面就是一个简单的血统图,优势:知道数据的操作记录,如果其中某一步骤的RDD丢失了,那么可根据血统关系图知道数据是怎么来的,可正向可反向,进而可恢复数据。
7,延迟计算(lazy Evaluation):spark对RDDs的计算,是他们第一次使用action操作时,通俗说就是只有在数据被必要使用是才去加载,同java的懒加载,比如我们使用transformation对数据转换,但是如果到最后我们并没有使用转换后的数据区计算结果呢?这样岂不是白白耗费资源了吗?这在大数据中是十分显著的,那么怎么知道在使用时,再去执行呢?Spark内部有metadata表会记录转换的操作记录。
五、RDD操作函数分Transformation和Action两类:
(1)Transformation是转换的意思,顾名思义就是把数据从一种形式转变成另一种形式,可以理解成转成方便我们看,好比把一长串的字符串,转成json树状图。
(2)Action是执行的意思,spark提供了很多算子,伴随DAG图。
(3)两个可以理解成对应hadoop中的map和reduce操作。
(4)没有action操作,单单转换是没有意义的。
六、Spark并行化就是执行了parallize()方法,如:sc.parallize(arrry)
七、sparkContext是一个对象,代表和一个集群的链接。
Sc.textFile()即是加载对象。
八、再理解下shuffle过程:把不同partition下相同的key聚集到一个partition下,造成了数据在内存中的重新分布。这也就是所谓的打乱、洗牌。
Shuffle过程分两个阶段:shuffle write和shuffle fetch。Shuffle write将shuffle MapTask任务产生的中间结果缓存到内存中,shuffle fetch获取shuffleMapTask缓存的中间结果进行shuffleReduceTask计算。
发布者:全栈程序员-用户IM,转载请注明出处:https://javaforall.cn/106043.html原文链接:https://javaforall.cn
【正版授权,激活自己账号】: Jetbrains全家桶Ide使用,1年售后保障,每天仅需1毛
【官方授权 正版激活】: 官方授权 正版激活 支持Jetbrains家族下所有IDE 使用个人JB账号...