1-3 - 爱码网

reduceByKey,aggregateByKey,combineByKey底层调用的都是combineByKeyWithClassTag

aggregate算子

def aggregate[U: Cla ssTag](zeroValue: U)(seqOp: (U, T) => U, combOp: (U, U) => U): U={}
zeroValue:初始值
seqOp:局部聚合
conbOp:全局聚合
初始值作用于聚不聚和和全局聚合

repartition, coalesce, partitionBy算子

会进行重新分区，是一种优化方式，解决数据倾斜问题
1.repartition肯定会发生shuffle，调用coalesce，传入true

def repartition(numPartitions: Int)(implicit ord: Ordering[T] = null): RDD[T] = withScope {
    coalesce(numPartitions, shuffle = true)//传入true，进行shuffle
  }

2.coalesce可以发生shuffle，也可以不发生，第二个参数由自己传入

coalesce(numPartitions: Int, shuffle: Boolean = false,
               partitionCoalescer: Option[PartitionCoalescer] = Option.empty)
              (implicit ord: Ordering[T] = null)

3.partitionBy用在自定义分区器的情况下

 def partitionBy(partitioner: Partitioner): RDD[(K, V)]

checkpoint

恢复的应用场景，防止数据丢失,tansformation类型，懒加载,以二进制方式存储

 sc.setCheckpointDir("hdfs://hadoop01:8020/cp")
val rdd = sc.textFile("hdfs://hadoop01:8020/wc").flatMap(_.split(" ")).map((_, 1)).reduceByKey(_+_)
rdd.checkpoint
rdd.isCheckpointed
rdd.count //action类型算子，执行之前的transformation类型的算子
rdd.isCheckpointed //是否做checkpoint
rdd.getCheckpointFile //获取rdd检查点路径

Spark内部通信协议

spark1.3(不包括1.3) actor
spark1.3-1.5 akka
spark1.6 akka+netty
spark2.0 netty

spark1.3(不包括1.3)	actor
spark1.3-1.5	akka
spark1.6	akka+netty
spark2.0	netty

Spark启动集群，节点之间的通信

Worker：进行计算的节点

集群启动步骤

调用start-all.sh启动集群，首先启动Master服务
Master服务启动后，会调用定时器，定时检查超时的Worker
通过解析slaves配置文件，开始在相应的节点启动Worker服务
Worker服务启动后，开始向Master发送注册信息
Master收到注册信息并保存，开始向Worker相应注册成功的信息（masterUrl）
worker收到masterUrl后并保存，开始向Master发送心跳

Master启动后：
1.检查Worker，有哪些Worker宕掉了，内部有定时器
2.等待Worker向他发送注册信息
Worker：
1.向Master进行注册，Master响应注册信息（就是发送Master自己的Url）
2.启动心跳向Master进行心跳汇报

Spark任务提交流程

1-3

Driver向Master进行注册（发送任务描述信息）
Master收到任务描述后，开始生成任务信息，并放到队列中
Master开始通知Worker启动Executor并发送任务信息
Worker接收到任务信息，开始启动Executor
Executor向Driver注册
Driver把生成的Task发送给相应的Executor

Driver：
spark-submit
–master hdfs://hadoop01:7077
–executor-memory 1024m
–total-executor-cores 2
/jarpath
args

Worker收到Master的通知启动Executor后，启动Executor，紧接着Executor向Driver进行注册，Driver将Task发送给相应的Executor