7.Spark Streaming
Spark Streaming是Spark核心api的一个拓展，可以实现高吞吐量/具备容错机制的实时流数据的处理
Spark Streaming 与 Spark Core 的关系可以用下面的经典部件图来表述：
images
基于Spark做Spark Streaming的思路
第一步
-
假设我们有一小块数据，那么通过RDD Api，我们能够构造出一个进行数据处理的RDD DAG
images
第二步
- 我们对连续的Streaming data进行切片处理，比如将最近200ms时间的event积攒一下，每个切片就是一个batch，然后使用第一步中的RDD DAG对这个batch的数据进行处理
注意：这里使用的是batch概念，确实200ms在其他同类系统中通常叫做mini-batch,不过既然Spark Streaming官方的叫法就是batch,我们这里就用batch表达mini-batch的意思了
-
针对连续不断的Streaming data进行多次切片，就会形成多个batch，也就对应出来多个RDD DAG（每个RDD DAG针对一个batch的数据），如此一来，这多个RDD DAG之间相互同构，却又是不同的实例，如图：
images
-
我们需要做的是
1.一个静态的RDD DAG的模版来表示处理逻辑
2.一个动态的工作控制器，将连续的Streaming data切分数据片段，并按照模版复制出新的RDD DAG的实例，对数据分段进行处理
images
第三步
- Hadoop MapReduce，Spark RDD api进行批处理时，一般默认数据已经在HDFS，HBase或其他存储上。而Streaming data，比如Twitter流，又有可能是在系统外实时产生的，就需要能够将这些数据导入到Spark Streaming系统中，类似Storm中的Spout
-
Spark Streaming支持多种数据源获取数据
images
-
我们需要做的是
3.原始数据的产生和导入
第四步
- Streaming job的运行时间是正无穷大的
-
我们需要做的是
4.对长时运行任务的保障，包括输入数据的失效后的重构，处理任务的失败后的重调
总结
Streaming data的特点决定了，如果我们想基于Spark Core进行Streaming data的处理，还需要在Spark Core的框架上解决刚才列出的1.2.3.4.的问题
images
Spark Streaming的整体模块划分
images
模块1：DAG静态定义
1.铺垫
- 首先对计算逻辑描述为一个RDD DAG的“模版”，在后面的job动态生成的时候，针对每个batch，Spark Streaming都将根据这个“模版”生成一个RDD DAG的实例
- 在Spark Streaming中，这个RDD“模版”对应的具体的类是DStream，RDD DAG“模版”对应的具体的类是DStreamGraph
- RDD本身也有很多子类，几乎每个子类都有一个对应的DStream，如UnionRDD对应的是UnionDStream。RDD通过transformation连接成RDD DAG(但RDD DAG在Spark Core里没有对应的具体类),DStream也通过transformation连接成DStreamGraph
2.DStream
- 定义：Spark Streaming提供了表示连续数据流的、高度抽象的被称为离散流
全限定名：org.apache.spark.streaming.dstream.DStream
- 做什么：将连续的数据持久化、离散化，然后进行批量处理
- 数据持久化：接收到的数据暂存
- 离散化：按时间分片，形成处理单元
- 分批处理
images
- 作用在DStream上的Operation分成两类
- Transformation：转换算子
- Spark支持RDD进行各种转换，因为DStream是由RDD组成的，Spark Streaming提供了一个可以在DStream上使用的转换集合，这些集合和RDD上可用的转换类似
- Spark Streaming提供了reduce和count这样的算子，但不会直接触发DStream计算
- 常用算子：Map，flatMap，join，reduceByKey
- OutPut：执行算子或输出算子
- print：控制台输出
- saveAsObjectFile,saveAsTextFile,saveAsHadoopFiles:将一批数据输出到Hadoop文件系统中,用批量数据的开始时间戳来命名
- forEachRDD：允许用户对DStream的每一批量数据对应的RDD本身做任意操作
- Transformation：转换算子
- DStream和RDD的关系
- DStream内部包含了多个RDD，代表了一系列的连续的RDD，每一个RDD包含特定的时间间隔
-
DStream维护了对每个产出的RDD实例的引用，如图中，DStream在3个batch里分别实例化了3个RDD，分别是a[1],a[2],a[3],那么 DStream A 就保留了一个 batch → 所产出的 RDD 的哈希表，即包含 batch 1 → a[1], batch 2 → a[2], batch 3 → a[3] 这 3 项
images
- 另外，能够进行流量控制的DStream子类，如ReceiverInputDStream，还会保存关于历次batch的源头数据条数，历次batch计算花费的时间等数值，用来实时计算准确的流量控制信息，这些都是记在DStream里的，而RDD a[1]等则不会保存这些信息
- 和Storm对比
- 在DStreamGraph的图里，DStream（即数据）是顶点，DStre之间的transformation（即计算）是边，这与Apache Storm是相反的
- 在Apache Storm的Topology里，计算是顶点，stream（连续的tuple，即数据）是边
-
DStream即是数据本身，在有向图里是顶点，而不是边
images
3.DStreamGraph
全限定名：org.apache.spark.streaming.DStreamGraph
-
定义：一系列transformation操作的抽象
- 示例：c = a.join(b), d = c.filter() 时， 它们的 DAG 逻辑关系是a/b → c，c → d，但在 Spark Streaming 在进行物理记录时却是反向的 a/b ← c, c ← d，目的为了追溯
images
- 示例：c = a.join(b), d = c.filter() 时， 它们的 DAG 逻辑关系是a/b → c，c → d，但在 Spark Streaming 在进行物理记录时却是反向的 a/b ← c, c ← d，目的为了追溯
-
DStream之间的转换锁形成的依赖关系全部保存在DStreamGraph中，DStreamGraph对于后期生成RDD Graph至关重要
-
DStreamGraph有点像简洁版的DAG schedule，负责根据某个时间间隔生成一系列JobSet以及按照依赖关系序列化
images
images
模块2：Job动态生成
- 在Spark Streaming程序的入口，我们都会定义一个batchDuration，就是需要每隔多长时间就比照静态的DStreamGraph来动态生成一个RDD DAG，在Spark Streaming里，总体负责动态作业调度的具体类是JobScheduler，在Spark Streaming程序开始运行的时候，会生成一个JobScheduler的实例，并被start()运行起来
-
JobScheduler有两个非常重要的成员：JobGenerator和ReceiverTracker。JobScheduler将每隔batch的RDD DAG具体生成工作委托给JobGenerator，而将源头输入数据的记录工作委托给ReceiveTracker
images
全限定名
- JobScheduler 的全限定名是：org.apache.spark.streaming.scheduler.JobScheduler
- JobGenerator 的全限定名是：org.apache.spark.streaming.scheduler.JobGenerator
- ReceiverTracker 的全限定名是：org.apache.spark.streaming.scheduler.ReceiverTracker
JobGenerator
JobGenerator维护了一个定时器，周期就是我们刚刚提到的batchDuration，定时为每个batch生成RDD DAG的实例。具体的，每次RDD DAG实际生成包含5个步骤
1.要求ReceiverTracker将目前已收到的数据进行一次allocate，即将上次batch切分后的数据切分到本次新的batch里
2.要求DStreamGraph复制出一套新的RDD DAG的实例，具体过程是：DStreamGraph将要求图例的尾DStream节点生成具体的RDD实例，并递归的调用尾DStream的上游DStream节点。以此遍历整个DStreamGraph，遍历结束也就正好生成了RDD DAG的实例
3.获取第1步ReceiverTracker分配到本batch的源头数据的meta信息
4.将第2步生成的本batch的RDD DAG，和第3步获取到的meta信息，一同提交给JobScheduler异步执行
5.只要提交结束（不管是否已经开始异步执行），就马上对整个系统的当前运行状态做一个checkpoint
调用关系如图
images
模块3：数据产生和导入
DStream有一个重要而特殊的子类ReceiverInputDStream：它除了需要像其他DStream那样在某个batch实例化RDD以外，还需要额外的Receiver为这个RDD生产数据
流程
1.由Receiver的总指挥ReceiverTracker分布多个job（每个job有1个task），到多个executor上分别启动ReceiverSupervisor实例
2.每个ReceiverSupervisor启动后将马上生成一个用户提供的Receiver实现的实例，该Receiver实现可以持续产生或者持续接收系统外数据，比如TwitterReceiver可以实时爬取Twitter数据，并在Receiver实例生成后调用Receiver.onStart()
1,2步骤的流程图如下，这时Receiver启动工作已经运行完毕，接下来ReceiverSupervisor将在executor端作为主要角色
images
ReceiverSupervisor 的全限定名是：org.apache.spark.streaming.receiver.ReceiverSupervisor
Receiver 的全限定名是：org.apache.spark.streaming.receiver.Receiver
3.Receiver在onStart()启动后，就将持续不断的接收外界数据，并持续交给ReceiverSupervisor进行数据转储
4.ReceiverSupervisor持续不断的接收到Receiver传来的数据：
- 如果数据很细小，就需要BlockGenerator攒多条数据成一块（4a），然后再成块存储（4b或4c）
- 反之就不用攒，直接成块存储（4b或4c）
- 这里Spark Streaming目前支持两种成块存储方式，一种是由 BlockManagerBasedBlockHandler 直接存到 executor 的内存或硬盘，另一种由 WriteAheadLogBasedBlockHandler 是同时写 WAL(4c) 和 executor 的内存或硬盘
5.每次成块在executor存储完毕后，ReceiverSupervisor就会及时上报块数据的meta信息给driver端的ReceiverTracker
这里的meta信息包括数据的标识id，数据的位置，数据的条数，数据的大小等信息
6.ReceiverTracker再将收到的块数据meta信息直接转给自己的成员ReceivedBlockTracker，由ReceivedBlockTracker专门管理收到的块数据meta信息
3,4,5,6的调用图如下，3,4,5,6的过程是一直持续不断的发生的
images
后续在driver端，就由ReceiverInputDStream在每个batch去检查ReceiverTracker收到的块meta信息，界定哪些新数据需要在本batch内处理，然后生成相应的RDD实例去处理这些块数据，这个过程在模块 1：DAG 静态定义 模块2：Job 动态生成 里描述过了
模块4：长时容错
Spark Streaming的长时容错特性，能够提供不重，不丢，exactly-once的处理语义
executor端
- 在executor端，ReceiverSupervisor和Receiver失效后直接重启就ok了，关键是保障收到的块数据的安全。保障了源头块数据，就能够保障RDD DAG（Spark Core的lineage）重做
- Spark Streaming对源头块数据的保障，分为4个层次，全面、相互补充，又可根据不同场景灵活配置：
1.热备：热备是指在存储块数据时，将其存储到本executor，并同时replicate到另外一个executor上去。这样在一个replica失效后，可以立刻无感知切换到另一份replica进行计算。实现方式是，在实现自己的Receiver时，即指定一下StorageLevel为 MEMORY_ONLY_2 或 MEMORY_AND_DISK_2 就可以了。
2.冷备：冷备是每次存储块数据前，先把块数据作为log写出到WriteAheadLog里，再存储到本executor。executor失效时，就由另外的executor去读WAL，再重做log来恢复数据块。WAL通常写到可靠存储如HDFS上，所以恢复时可能需要一段recover time
images
3.重放：如果上游支持重放，比如Apache Kafka，那么就可以选择不用热备或者冷备来另外存储数据了，而是在失效时换一个executor进行数据重放即可。
4.忽略：最后，如果应用的实时性需求大于准确性，那么一块数据丢失后我们可以选择忽略、不恢复失效的源头数据。
总结
images
driver端
- 块数据的meta信息上报到ReceiverTracker，然后交给ReceivedBlockTracker做具体的管理。ReceivedBlockTracker也采用WAL冷备方式进行备份，在driver失效后，由新的ReceivedBlockTracker读取WAL并恢复block的meta信息
- 需要定时对DStreamGraph和JobScheduler做Checkpoint，来记录整个DStreamGraph的变化和每个batch的job的完成情况
注意到这里采用的是完整checkpoint的方式，和之前的WAL的方式都不一样。checkpoint通常也是落地到可靠存储如HDFS。checkpoint发起的间隔默认的是和batchDuration一致；即每次batch发起，提交了需要运行的job后就做checkpoint，另外在job完成了更新任务状态的时候再次做一下checkpoint
这样一来，在driver失效并恢复后，可以读取最近一次的checkpoint来恢复作业的DStreamGraph和job的运行及完成状态
总结
images
架构简析
组成
- master：记录DStream之间的依赖关系或者血缘关系，并负责任务调度以生成新的RDD
- worker：从网络接收数据，存储并执行RDD计算
- 处理数据模式
1.recevier模式：被动，异步
优点：快
缺点：启动多个executor，至少需要2个线程才行
2.direct模式：主动，同步
优点：一个executor占用资源少
缺点：慢
- 处理数据模式
-
client：负责向Spark Streaming中灌入数据
images
作业提交
-
Network Input Tracker
跟踪每一个网络received数据，并且将其映射到相应的input DStream上 -
Job Scheduler
周期性访问DStream Graph并生成Spark Job，将其交给Job Manager执行 -
Job Manager
获取任务队列，并执行Spark任务
images
窗口操作
Spark提供了一组窗口操作，通过滑动窗口技术对大规模数据的增量更新进行统计分析
-
Window Operation:定时进行一定时间段内的数据处理
images
- 任何基于窗口操作需要指定两个参数，窗口总长度和滑动时间间隔
images
WAL容错
工作原理和恢复看图加上注释应该比较好理解
工作原理
images
恢复
images
基于Spark Streaming有上述蛮多内容的理解，部分内容是学习借鉴腾讯广告团队分享的内容。有理解错误的地方，欢迎大家积极指正。下面这个链接是腾讯广告团队的分享，大家可以学习学习。
