11.Spark Streaming源码解读之Driver中的ReceiverTracker架构设计以及具体实现彻底研究

上篇文章详细解析了Receiver不断接收数据的过程,在Receiver接收数据的过程中会将数据的元信息发送给ReceiverTracker:

本文将详细解析ReceiverTracker的的架构设计和具体实现

一、ReceiverTracker的主要功能

ReceiverTracker的主要功能有:

1.在Executor上启动Receivers

2.接受Receiver的注册

3.借助ReceivedBlockTracker来管理Receiver接收数据的元数据

4.接受Receiver发送的各种消息,并作相应处理

5.更新Receiver接收数据的速率(也就是限流)

6.不断的等待Receivers的运行状态,只要Receivers停止运行,就重新启动Receiver。也就是Receiver的容错功能。

7.停止Receivers

8.汇报Receiver发送过来的错误信息

二、ReceiverTracker具体功能详解

2.1 启动receiver并管理receiver接收数据的元数据

 

首先,ReceiverTracker内部有一个ReceiverTrackerEndPoint通讯体endpoint变量,endpoint用来和Receiver和ReceiverTracker本身进行消息通讯。这个ReceiverTrackerEndPoint通讯体在ReceiverTracker启动时被初始化:

ReceiverTracker启动Receiver时候,向ReceiverTrackerEndPoint通讯体endpoint变量发送了StartAllReceivers(receivers)消息:

Receiver启动后会向ReceiverTracker注册,告诉ReceiverTracker自己启动成功:

代码中的trackerEndpoint就是ReceiverTracker中ReceiverTrackerEndPoint通讯体endpoint的引用。

Receiver会不断将接收的数据封装成Block,并将这些Block推送给BlockManager管理,在将这些Block推送给BlockManager之后,ReceiverSupervisor会将Block的元信息发送给ReceiverTracker的endpoint:

可以看到ReceiverSupervisor向ReceiverTracker的endpoint发送了AddBlock(blockInfo)消息:

ReceiverTracker收到AddBlock(blockInfo)消息后,会启动一个线程进行处理:

ReceiverTracker收到AddBlock(blockInfo)消息后,调用了addBlock(receiveedBlockInfo)方法进行处理,下面是addBlock的源码:

这里其实调用了receivedBlockTracker的addBlock方法,receivedBlockTracker是ReceivedBlockTracker对象,它是在ReceiverTracker实例化时候被创建:

下面看一下ReceivedBlockTracker的addBlock方法:

可以看到ReceivedBlockTracker的addBlock方法将block的元信息添加到了一个队队列中,最终是添加到一个叫做streamIdToUnallocatedBlockQueues的HashMap中,其中key是streamId,值是该streamid对应的block队列。

2.2 为Batch分配Block

当spark streaming应用程序动态生成job的时候,JobGenerator会调用generateJobs方法,在该方法中会为批处理分配已经接收的Block

这里调用了jobScheduler中receiverTracker的allocatedBlockToBatch方法,这里的receiverTracker就是ReceiverTracker对象,下面看一下该方法的实现:

可以看到,最终调用了ReceivedBlockTracker的allocatedBlockToBatch方法:

这里先根据streamId,从streamIdToUnallocatedBlockQueues中取出接收到的block队列,并将streamId和block队列封装成AllocatedBlocks,最后根据batchTime将其对应的AllocatedBlocks对象加入timeToAllocatedBlocks中,timeToAllocatedBlocks是一个HashMap:

这样Batch的Block就分配完成。

2.3 ReceiverTracker处理的其他消息

ReceiverTracker中ReceiverTrackerEndpoint的receive方法定义了各种消息的处理逻辑:  

(1) 收到StartAllReceivers(receivers)消息后,ReceiverTracker会为receivers分配executor,并在executor上启动相应的receiver

(2)当ReceiverTracker监控到receiver退出返回时,会给ReceiverTrackerEndpoint发送RestartTracker(receiver)消息。收到该消息后,会重新为receiver分配executor启动receiver(如果原来的executor运行正常就在原先的executor上重新启动,否则重新调度executor)。

(3)当Spark Streaming 的job结束后,JobScheduler会调用handleJobCompletion方法,最终会调用cleanupOldBlocksAndBatches方法给endpoint发送CleanupOldBlocks消息:

收到该消息后,会被路由到Receiver 进行Block的清理。

(4)UpdateReceiverRateLimit消息

 

收到UpdateReceiverRateLimit消息后,会将其路由到receiver,当receiver收到该消息后会调用BlockGenerator的update方法更新Block生成速率。

From WizNote

时间: 2024-10-27 18:35:05

11.Spark Streaming源码解读之Driver中的ReceiverTracker架构设计以及具体实现彻底研究的相关文章

第11课:Spark Streaming源码解读之Driver中的ReceiverTracker架构设计以及具体实现彻底研究

本篇博文的目标是: Driver的ReceiverTracker接收到数据之后,下一步对数据是如何进行管理 一:ReceiverTracker的架构设计 1. Driver在Executor启动Receiver方式,每个Receiver都封装成一个Task,此时一个Job中就一个Task,而Task中就一条数据,也就是Receiver数据.由此,多少个Job也就可以启动多少个Receiver. 2. ReceiverTracker在启动Receiver的时候他有ReceiverSuperviso

(版本定制)第11课:Spark Streaming源码解读之Driver中的ReceiverTracker彻底研究和思考

本期内容: 1.ReceiverTracker的架构设计 2.消息循环系统 3.ReceiverTracker具体实现 上节课讲到了Receiver是如何不断的接收数据的,并且接收到的数据的元数据会汇报给ReceiverTracker,下面我们看看ReceiverTracker具体的功能及实现. ReceiverTracker主要的功能: 在Executor上启动Receivers. 停止Receivers . 更新Receiver接收数据的速度(也就是限流) 不断的等待Receivers的运行

Spark Streaming源码解读之Driver中ReceiverTracker架构设计以具体实现彻底研究

本期内容 : ReceiverTracker的架构设计 消息循环系统 ReceiverTracker具体实现 一. ReceiverTracker的架构设计 1. ReceiverTracker可以以Driver中的具体自己的算法来在具体的Execute中启动Receiver,启动Receiver的方式会把每个Receiver都封装成为一个Tracker, Tracker是这个Job中唯一的Tracker,实质上讲ReceiverTracker启动Receiver的方式就是封装成一个个Job ,

(版本定制)第13课:Spark Streaming源码解读之Driver容错安全性

本期内容:1. ReceiverBlockTracker容错安全性 2. DStream和JobGenerator容错安全性 一:容错安全性 1. ReceivedBlockTracker负责管理Spark Streaming运行程序的元数据.数据层面 2. DStream和JobGenerator是作业调度的核心层面,也就是具体调度到什么程度了,从运行的考虑的.DStream是逻辑层面. 3. 作业生存层面,JobGenerator是Job调度层面,具体调度到什么程度了.从运行的角度的. 谈D

Spark发行版笔记13:Spark Streaming源码解读之Driver容错安全性

本节的主要内容: 一.ReceivedBlockTracker容错安全性 二.DStreamGraph和JobGenerator容错安全性 从数据层面,ReceivedBlockTracker为整个Spark Streaming应用程序记录元数据信息. 从调度层面,DStreamGraph和JobGenerator是Spark Streaming调度的核心,记录当前调度到哪一进度,和业务有关. ReceivedBlockTracker在接收到元数据信息后调用addBlock方法,先写入磁盘中,然

第13课:Spark Streaming源码解读之Driver容错安全性

本期内容: ReceivedBlockTracker容错安全性 DStream和JobGenerator容错安全性 Driver的容错有两个层面:1. Receiver接收数据的元数据 2. Driver管理的各组件信息(调度和驱动层面) 元数据采用了WAL的容错机制 case AddBlock(receivedBlockInfo) =>   if (WriteAheadLogUtils.isBatchingEnabled(ssc.conf, isDriver = true)) {     wa

Spark 定制版:009~Spark Streaming源码解读之Receiver在Driver的精妙实现全生命周期彻底研究和思考

本讲内容: a. Receiver启动的方式设想 b. Receiver启动源码彻底分析 注:本讲内容基于Spark 1.6.1版本(在2016年5月来说是Spark最新版本)讲解. 上节回顾 上一讲中,我们给大家具体分析了RDD的物理生成和逻辑生成过程,彻底明白DStream和RDD之间的关系,及其内部其他有关类的具体依赖等信息: a. DStream是RDD的模板,其内部generatedRDDs 保存了每个BatchDuration时间生成的RDD对象实例.DStream的依赖构成了RDD

Spark发行版笔记10:Spark Streaming源码解读之流数据不断接收和全生命周期彻底研究和思考

本节的主要内容: 一.数据接受架构和设计模式 二.接受数据的源码解读 Spark Streaming不断持续的接收数据,具有Receiver的Spark 应用程序的考虑. Receiver和Driver在不同进程,Receiver接收数据后要不断给Deriver汇报. 因为Driver负责调度,Receiver接收的数据如果不汇报给Deriver,Deriver调度时不会把接收的数据计算入调度系统中(如:数据ID,Block分片). 思考Spark Streaming接收数据: 不断有循环器接收

15、Spark Streaming源码解读之No Receivers彻底思考

在前几期文章里讲了带Receiver的Spark Streaming 应用的相关源码解读,但是现在开发Spark Streaming的应用越来越多的采用No Receivers(Direct Approach)的方式,No Receiver的方式的优势: 1. 更强的控制自由度 2. 语义一致性 其实No Receivers的方式更符合我们读取数据,操作数据的思路的.因为Spark 本身是一个计算框架,他底层会有数据来源,如果没有Receivers,我们直接操作数据来源,这其实是一种更自然的方式