day1112
1.spark core复习
任务提交
缓存
checkPoint
自定义排序
自定义分区器
自定义累加器
广播变量
Spark Shuffle过程
SparkSQL
一. Spark Core回顾
1 集群启动
Spark集群启动流程:
1、调用start-alsh脚本,开始启动Master
2、Master启动以后, preStart方法调用了一个定时器,定时的检查超时的Worker.
3、启动脚本会解析slaves配置文件,找到启动Worker的相应节点,开始启动Worker
4、Worker服务启动后开始调用preStart方法开始向所有的Master进行注册
5、Master接收到Worker发送过来注册信息, Master开始保存注册信息并把自己的url响应给Worker
6、Worker接收到Master的url后并更新 ,开始调用一个定时器 ,定时的向Master发送心跳信息
7、Master不断的收到心跳信息(WorkerId),Master根据WorkerId去找是否有之前存过的WorkerId对应的WorkerInfo,如果有则更新最后一次心跳信息
2 任务提交
Spark任务提交流程:
1、Driver端会通过spark-sumbit脚本启动SparkSubmit进程,此时创建了一个非常重要的对象
(SparkContext),开始向Master发送信息
2、Master接收到发送过来的信息后开始生成任务信息,并把任务信息放到一个队列里
3、Master把所有有效的Worker过滤出来,按照空闲的资源进行排序
4、Master开始向有效的Worker通知拿取任务信息并启动相应的Executor
5、Worker启动Executor并向Driver反向注册
6、Driver开始把生成的task发送给相应的Executor,Executor开始执行任务
任务提交划分四个阶段:
->生成RDD的过程
->stage划分过程(DAGScheduler)
-> task生成和调度过程(TaskScheduler)
->Executor执行task过程(在Executor端产生)
3 缓存
应用场景:接下来的操作会经常用到某个RDD,该RDD最好进行缓存处理
调用方法:cache,persist
可以指定缓存级别
若要存4G,但是只能存2G:按照先存2G.剩余的不存,调用时没有存进去的重新计算,存进去的直接调用
4 checkPoint
存储的路径:HDFS
应用场景:为了缩短依赖链条,起到数据容错,最终还是为了提高计算效率,一般在重要的shuffle后需要进行checkpoint
操作步骤:
|->1:设置checkpoint目录
|->2:将RDD进行cache
|->3:将RDD进行checkpoint
5 自定义排序
应用场景:为了解决比较复杂的排序需求,(例如:自定义一个对象,里面有多个字段,先按照某个字段进行排序,如果相同,接下来在按照另一个字段进行排序)
6 自定义分区器
应用场景:Spark的分区器无法实现的业务场景,可以用自定义分区器实现
实现:需要继承Partition抽象类,实现getPartition,numPartitions方法
7 自定义累加器
应用场景:有累加的需求,需要将累加的结果赋值给Driver端的某个变量
注意:累加的变量只能在Driver端,而且是由Executor的task进行累加(分布式累加),Executor端无法读取累加值
8 广播变量
应用场景:Driver端的某个变量的值在分布式计算(task的执行过程中)过程中会经常用到该值,为了提高计算效率,可以用广播变量的方式将变量的值广播到相应的Executor中
注意:
|->1:广播变量不一定会对每个Executor都发送一遍,只是会将变量的值发送给每个节点一次,该节点其他Executor用该变量只会从本节点的Executor取数据,不会从Driver端去拿.如果本节点没有,则遵循就近原则,从离他最近的节点取值.
|->2:广播出去的值不可以后期更改,RDD是无法广播的.
9 Spark Shuffle过程(重要)
shuffle分maptask和reducetask,shuffle过程是跨stage的
maptask端发生的是shufflewrite过程
reducetask端发生的shuffleread过程
10 SparkSQL
初始化兑现g:SparkSession
操作的对象:DataFrame,Dataset
DataFrame = Dataset[Row]:是Dataset的弱类型(自动推断类型),Dataset是强类型(开始就需要指定一个类型)
RDD,DataFrame,Dataset互相转化
两种操作方式:
|->DSL语言风格:select,filter,where,group by, join, agg(max/min/avg/count...)...
|->SQL语言风格:处理过程:
|->1.生成DataFrame/Dataset
|->2.注册临时表
|->3.调用上下文的sql方法