SparkStreaming(源码阅读十二)

简介:   要完整去学习spark源码是一件非常不容易的事情,但是咱可以积少成多嘛~那么,Spark Streaming是怎么搞的呢?  本质上,SparkStreaming接收实时输入数据流并将它们按批次划分,然后交给Spark引擎处理生成按照批次划分的结果流:    SparkStreaming提供了表示连续数据流的、高度抽象的被称为离散流的Dstream,可以使用kafka、Flume和Kiness这些数据源的输入数据流创建Dstream,也可以在其他Dstream上使用map、reduce、join、window等操作创建Dsteram。

  要完整去学习spark源码是一件非常不容易的事情,但是咱可以积少成多嘛~那么,Spark Streaming是怎么搞的呢?

  本质上,SparkStreaming接收实时输入数据流并将它们按批次划分,然后交给Spark引擎处理生成按照批次划分的结果流

  

  SparkStreaming提供了表示连续数据流的、高度抽象的被称为离散流的Dstream,可以使用kafka、Flume和Kiness这些数据源的输入数据流创建Dstream,也可以在其他Dstream上使用map、reduce、join、window等操作创建Dsteram。Dstream本质上呢,是表示RDD的序列

  Spark Streaming首先将数据切分为一定时间范围(Duration)的数据集,然后积累一批(Batch)Duration数据集后单独启动一个任务线程处理。Spark核心提供的从DAG重新调度任务和并行执行,能够快速完成数据从故障中恢复的工作。

  那么下来就从SparkStreaming 的StreamingContext初始化开始:

  StreamingContext传入的参数:1、SparkContext也就是说Spark Streaming的最终处理实际是交给SparkContext。2、Checkpoint:检查点.3、Duration:设定streaming每个批次的积累时间。当然,也可以不用设置检查点。

  Dstream是Spark Streaming中所有数据流的抽象,这里对抽象类Dstream定义的一些主要方法:

  1、dependencies:Dstream依赖的父级Dstream列表。

  2、comput(validTime:Time):指定时间生成一个RDD。

  3、isInitialized:Dstream是否已经初始化。

  4、persist(level:StorageLevel):使用指定的存储级别持久化Dstream的RDD。

  5、persist:存储到内存

  6、cache:缓存到内存,与persisit方法一样。

  (这里详细说下cache与persist的不同点:cache只有一个默认的缓存级别MEMORY_ONLY ,而persist可以根据情况设置其它的缓存级别。)

  7、checkpoint(interval:Duration):设置Dstream及祖宗Dstream的DstreamGraph;

  8、getOrCompute(time:Time):从缓存generatedRDDs = new HashMap[Time,RDD[T]]中获取RDD,如果缓存不存在,则生成RDD并持久化、设置检查点放入缓存。

  9、generateJob(time:Time):给指定的Time对象生成Job.

  10、window(windowDuration:Duration):基于原有的Dstream,返回一个包含了所有在时间滑动窗口中可见元素的新的Dstream.

  ......

  Dsteam本质上是表示连续的一些列的RDD,Dstream中的每个RDD包含了一定间隔的数据,任何对Dstream的操作都会转化为底层RDD的操作。在Spark Streaming中,Dstream提供的接口与RDD提供的接口非常相似。构建完ReciverInputDStream后,会调用各种Dstream的接口方法,对Dstream进行各种转换,最后各个Dstream之间的依赖关系就形成了一张DStream Graph:

  整个流程所涉及的组件为:

  1、Reciever:Spark Streaming内置的输入流接收器或用户自定义的接收器,用于从数据源接收源源不断的数据流。

  2、currentBuffer:用于缓存输入流接收器接收的数据流。

  3、blockIntervalTimer:一个定时器,用于将CurrentBuffer中缓存的数据流封装为Block后放入blocksForPushing。

  4、blockForPushing:用于缓存将要使用的Block。

  5、blockPushingThread:此线程每隔100毫秒从blocksForPushing中取出一个Block存入存储体系,并缓存到ReceivedBlockQueue。

  6、Block Batch:Block批次,按照批次时间间隔,从RecievedBlockQueue中获取一批Block。

  7、JobGenerator:Job生成器,用于给每一批Blcok生成一个Job。

   下来继续回到StreamingContext,在StreamingContext中new了一个JobScheduler,它里面创了EventLoop,对这个名字是不是很熟悉?没错,就是在Netty通信交互时创建的对象,主要用于处理JobSchedular的事件。然后启动StrreamingListenerBus,用于更新Spark UI中的StreamTab的内容。 那么最重要的就是下来创建ReceiverTracker,它用于处理数据接收、数据缓存、Block生成等工作。最后启动JobGenerator,负责对DstreamGraph的初始化Dstream与RDD的转换生成JOB提交执行等工作。

  

  曾经是用ReciverTrackerActor接收来自Reciver的消息,包括RegisterReceiver、AddBlock、ReportError、DeregisterReceiver等,现在不再使用Akka进行通信,而是使用RPC。

  回到launchReceivers,调用了SparkContext的makeRDD方法,将所有Receiver封装为ParallelCollectionRDD,并行度是receivers的数量,makeRDD方法实际调用了parallelize:

  

  今天到此为止。。明天再来会你这磨人的小妖精,玩别的去啦~~~

 

参考文献:《深入理解Spark:核心思想与源码分析》

目录
相关文章
|
网络协议 Shell Android开发
解决adb网络连接中出现的“由于目标计算机积极拒绝,无法连接”错误
解决adb网络连接中出现的“由于目标计算机积极拒绝,无法连接”错误
解决adb网络连接中出现的“由于目标计算机积极拒绝,无法连接”错误
|
SQL 人工智能 自然语言处理
一款利用人工智能将自然语言查询转换为 SQL 代码的互译工具 - SQL Translator
一款利用人工智能将自然语言查询转换为 SQL 代码的互译工具 - SQL Translator
363 0
|
8月前
|
弹性计算 安全 程序员
云服务诊断真实使用评测
云服务诊断真实使用评测
|
监控 算法 安全
转:文档管理系统中如何利用巴伐利亚算法实现高效使用
巴伐利亚算法(Bavarian Sketching)是一种基于哈希表的数据结构,可以高效地实现近似计数和查询。
131 0
|
JSON 数据格式 Python
Python处理数据json还有哪些基本操作?
Python处理数据json还有哪些基本操作?
161 0
|
XML API 数据格式
【Qt 学习笔记】QWidget的enable属性 | API的介绍
【Qt 学习笔记】QWidget的enable属性 | API的介绍
379 0
|
存储 安全 Java
Java集合
Java 集合主要包括 Collection 和 Map 两种: 1、Collection 是一种存储元素的集合,Collection 包括 List、Set、Queue 三种类型。 2、Map 是一种存储键值对的集合。
219 0
Java集合
|
Dubbo 应用服务中间件 开发者
启动检查|学习笔记
快速学习启动检查
启动检查|学习笔记
|
存储 消息中间件 SQL
接口的幂等性的多重考虑,你会了吗?
今天的主题:接口幂等性的解决方案。本来是想把对象的存储过程和内存布局肝出来的,但是临时产生了变化,哈哈,这部分内容我们留在下一期吧,有句话说的好,好事多磨,对吧。 在实际项目开发中接口是我们在开发中经常接触到的,而且是经常经常要写,每一个项目可能都会伴随着大量的接口开发,在涂鸦的这几个月,基本上就是在与接口作斗争了,新需求除了业务相关就是设计表和接口编写了。 当然,在接口设计中我们要考虑很多问题,安全性,格式,设计等等,今天我们先来聊聊,在高并发环境下,接口幂等性的解决方案有哪些。
接口的幂等性的多重考虑,你会了吗?
|
存储
存储基础知识 - 磁盘寻址(CHS寻址方式、LBA寻址方式)
http://blog.csdn.net/haiross/article/details/38659825
2014 0

热门文章

最新文章