20210207

Posted on 2021-06-13 Edited on 2021-06-14

大数据 | 大数据基础--系统之大数据实时计算框架-实例

亲爱的读者朋友大家晚上好，上次讲了实时计算框架的概述，这篇文章就紧接着上一篇文章具体看看storm和spark streaming的细节。

大数据的实时计算任务的抽象

需要解决的问题

操作的数据形式和结构是什么？
应用程序可以对于数据做何种操作？

需要考虑的问题

适用的场景/数据是什么？
面向的软硬件环境是什么？
和支撑程序程序之间的界面在哪里？

Storm

Storm处理对象的抽象

Streams：Storm将流数据Stream描述成一个无限的Tuple序列，这些Tuple序列会以分布式的方式并行地创建和处理
每个tuple是一堆值，每个值有一个名字，并且每个值可以是任何类型
Tuple应该是一个Key-Value的Map，由于各个组件间传递的tuple 的字段名称已经事先定义好了，所以Tuple只需要按序填入各个 Value，所以就是一个Value List（值列表）

流处理方式的抽象

Spout：Storm认为每个Stream都有一个源头，并把这个源头抽象为Spout
通常Spout会从外部数据源（队列、数据库等）读取数据，然后封装成Tuple形式，发送到Stream中。Spout是一个主动的角色，在接口内部有个nextTuple函数，Storm框架会不停的调用该函数

状态转换的抽象

Bolt：Storm将Streams的状态转换过程抽象为Bolt。Bolt即可以处理Tuple，也可以将处理后的Tuple作为新的Streams发送给其他Bolt
Bolt可以执行过滤、函数操作、Join、操作数据库等任何操作
Bolt是一个被动的角色，其接口中有一个execute(Tuple input)方法，在接收到消息之后会调用此函数，用户可以在此方法中执行自己的处理逻辑

操作关系的抽象

Topology：Storm将Spouts和Bolts组成的网络抽象成Topology，它可以被提交到Storm集群执行。Topology可视为流转换图，图中节点是一个Spout或Bolt，边则表示Bolt订阅了哪个Stream。当 Spout或者Bolt发送元组时，它会把元组发送到每个订阅了该 Stream的Bolt上进行处理
Topology里面的每个处理组件（Spout或Bolt）都包含处理逻辑，而组件之间的连接则表示数据流动的方向
Topology里面的每一个组件都是并行运行的
- 在Topology里面可以指定每个组件的并行度， Storm会在集群里面分配那么多的线程来同时计算
- 在Topology的具体实现上， Storm中的Topology定义仅仅是一些Thrift结构体（二进制高性能的通信中间件），支持各种编程语言进行定义

组件数据关系的抽象

Stream Groupings：Storm中的Stream Groupings用于告知 Topology如何在两个组件间（如Spout和Bolt之间，或者不同的 Bolt之间）进行Tuple的传送。每一个Spout和Bolt都可以有多个分布式任务，一个任务在什么时候、以什么方式发送Tuple就是由 Stream Groupings来决定的。

Stream Grouping的方式

(1)ShuffleGrouping：随机分组，随机分发Stream中的Tuple，保证每个 Bolt的Task接收Tuple数量大致一致

(2)FieldsGrouping：按照字段分组，保证相同字段的Tuple分配到同一个 Task中

(3)AllGrouping：广播发送，每一个Task都会收到所有的Tuple

(4)GlobalGrouping：全局分组，所有的Tuple都发送到同一个Task中

(5)NonGrouping：不分组，和ShuffleGrouping类似，当前Task的执行会和它的被订阅者在同一个线程中执行

(6)DirectGrouping：直接分组，直接指定由某个Task来执行Tuple的处理

Spark Streaming

离散流数据的处理

以一系列非常小的、确定的批处理作业的形式运行流计算 1612440579389

将实时流分成若干批，每批X秒
Spark将每批数据视为RDD，并使用 RDD操作处理它们
最后，分批返回RDD操作的处理结果
批处理大小低至½秒，延迟约为1秒
可在同一系统中结合批处理和流处理

Spark Streaming设计

DStream表示连续不断的数据流。在内部实现上，Spark Streaming的输入数据按照时间片（如1秒）分成一段一段的DStream，每一段数据转换为Spark中的RDD，并且对DStream的操作都最终转变为对相应的RDD的操作。

Spark Streaming的核心概念

DStream –表示数据流的RDDs序列
- Twitter, HDFS, Kafka, Flume, ZeroMQ, Akka Actor, TCP套间字
Transformations – 从一个Dstream修改数据以创建另一个DStream
- 标准的RDD操作 – map, countByValue, reduce, insert……
- 有状态操作 – window, countByValueAndWindow……
输出操作 –向外部实体传送数据
- saveAsHadoopFiles – 保存到HDFS
- foreach – 对每一批结果进行处理

大数据的实时计算框架的API

需要解决的问题

程序员如何基于框架编写应用程序？

需要考虑的问题

面向什么样的程序员？
开发的效率？
程序的易读性？
程序的美感？
是否符合传统开发的习惯？

Storm

Storm开发API

Java API
- Spout
  - nextTuple()：回调函数，循环触发
  - ack(id)：回调函数，消息成功处理时触发
  - fail(id)：回调函数，消息超时时触发
- Bolt
  - execute(Tuple input)：回调函数，数据触发
  - collector.emit(tuple)：通过collector向下游发送tuple
  - collector.ack(tuple)：通过collector确认已经成处理输入tuple

Spark Streaming

DStream的输入源

在Spark Streaming中所有的操作都是基于流的，而输入源是这一系列操作的起点。输入 DStreams 和 DStreams 接收的流都代表输入数据流的来源，在 Spark Streaming 提供两种内置数据流来源：
- 基础来源 – 在 StreamingContext API 中直接可用的来源。例如：文件系统、Socket（套接字）连接和 Akka actors。
- 高级来源 – 如 Kafka、Flume、Kinesis、Twitter 等，可以通过额外的实用工具类创建。

大数据的实时处理系统的系统架构

需要解决的问题

系统有哪些模块？
模块之间如何交互？

需要考虑的问题

有效配合硬件
扩展性高
效率高
扩展性好
开发容易
维护方便
升级简单

Storm

技术架构

关键组件

Storm集群采用“Master—Worker”的节点方式
Nimbus（主节点）：主节点通常运行一个后台程序 —— Nimbus，用于响应分布在集群中的节点，分配任务和监测故障。这个很类似于 Hadoop中的Job Tracker。
Supervisor（工作节点）：工作节点同样会运行一个后台程序 Supervisor，用于收听工作指派并基于要求运行工作进程。每个工作节点都是 topology中一个子集的实现。而Nimbus和 Supervisor之间的协调则通过Zookeeper系统或者集群。
Zookeeper： Zookeeper是完成Supervisor和Nimbus之间协调的服务。而应用程序实现实时的逻辑则被封装进Storm中的 “topology”。topology则是一组由Spouts（数据源）和Bolts （数据操作）通过Stream Groupings进行连接的图
Storm使用Zookeeper来作为分布式协调组件，负责Nimbus和多个 Supervisor之间的所有协调工作。借助于Zookeeper，若Nimbus 进程或Supervisor进程意外终止，重启时也能读取、恢复之前的状态并继续工作，使得Storm极其稳定

Spark Streaming

Spark-Streaming运行原理- Spark Streaming构架

大数据的实时处理系统的基本数据操作

需要解决的问题

包括哪些基本数据操作？
并行还是串行实现？
高效实现算法？

需要考虑的问题

有效支撑API
容易理解
扩展性高
效率高

Storm

关键组件

Stream Groupings: Stream Grouping定义了一个流在Bolt任务间该如何被切分。这里有Storm提供的6个Stream Grouping类型：

随机分组（Shuffle grouping）：随机分发 tuple到Bolt的任务，保证每个任务获得相等数量的tuple。
字段分组（Fields grouping）：根据指定字段分割数据流，并分组。例如，根据“user-id”字段，相同“user-id”的元组总是分发到同一个任务，不同“user-id”的元组可能分发到不同的任务。
全部分组（All grouping）：tuple被复制到bolt的所有任务。这种类型需要谨慎使用。
全局分组（Global grouping）：全部流都分配到 bolt的同一个任务。明确地说，是分配给ID最小的那个 task。
无分组（None grouping）：你不需要关心流是如何分组。目前，无分组等效于随机分组。但最终，Storm 将把无分组的Bolts放到Bolts或Spouts订阅它们的同一线程去执行（如果可能）。
直接分组（Direct grouping）：这是一个特别的分组类型。元组生产者决定tuple由哪个元组处理者任务接收。

Spark Streaming

同Spark Streaming

大数据的批处理系统中的流程生成

Storm流程由用户指定

Spark Streaming流程由用户指定

大数据的批处理系统中的流程调度

### 需要解决的问题

执行过程中如何执行操作？

需要考虑的问题

可扩展性高
效率高

Storm

计算模式

流式计算
持续计算

Spark Streaming

Spark Streaming设计

Spark Streaming的基本原理是将实时输入数据流以时间片（秒级）为单位进行拆分，然后经Spark引擎以类似批处理的方式处理每个时间片数据。

运行原理

计算流程
- Spark Streaming是将流式计算分解成一系列短小的批处理作业。这里的批处理引擎是Spark Core。
- 也就是把Spark Streaming的输入数据按照batch size（如1秒）分成一段一段的数据（Discretized Stream），每一段数据都转换成Spark中的RDD。
- 然后将Spark Streaming中对DStream的Transformation操作变为针对Spark 中对RDD的Transformation操作，将RDD经过操作变成中间结果保存在内存中。
- 整个流式计算根据业务的需求可以对中间的结果进行叠加或者存储到外部设备。
batch size
- 时间片或批处理时间间隔（ batch interval），这是人为地对流数据进行定量的标准，以时间片作为我们拆分流数据的依据。一个时间片的数据对应一个 RDD实例。

事务处理

Storm

任务级失败

Bolt任务crash引起的消息未被应答。此时，acker中所有与此 Bolt任务关联的消息都会因为超时而失败，对应的Spout的 fail方法将被调用。
acker任务失败。如果acker任务本身失败了，它在失败之前持有的所有消息都将超时而失败。Spout的fail方法将被调用。
Spout任务失败。在这种情况下，与Spout任务对接的外部设备(如MQ)负责消息的完整性。例如，当客户端异常时， kestrel队列会将处于pending状态的所有消息重新放回队列中。

任务槽(slot)故障务级失败

Worker 失败。每个 Worker 中包含数个 Bolt( 或 Spout) 任务。 Supervisor负责监控这些任务，当worker失败后会尝试在本机重启它，如果它在启动时连续失败了一定的次数，无法发送心跳信息到 Nimbus，Nimbus将在另一台主机上重新分配worker。
Supervisor失败。Supervisor 是无状态(所有的状态都保存在Zookeeper或者磁盘上)和快速失败(每当遇到任何意外的情况，进程自动毁灭)的，因此Supervisor的失败不会影响当前正在运行的任务，只要及时将他们重新启动即可。
Nimbus失败。Nimbus也是无状态和快速失败的，因此Nimbus的失败不会影响当前正在运行的任务，但是当Nimbus失败时，无法提交新的任务，只要及时将它重新启动即可。

集群节点(机器)故障

Storm集群中的节点故障。此时Nimbus会将此机器上所有正在运行的任务转移到其他可用的机器上运行。
Zookeeper集群中的节点故障。Zookeeper保证少于半数的机器宕机系统仍可正常运行，及时修复故障机器即可。

Nimbus节点故障

如果失去了Nimbus节点，Worker也会继续执行。另外，如果worker死亡，Supervisor也会继续重启他们。但是，没有 Nimbus，Worker不会在必要时(例如，失去一个Worker的主机)被安排到其他主机，客户端也无法提交任务。所以 Nimbus“在某种程度上”是单点故障。在实践中，这不是一个大问题，因为Nimbus守护进程死亡，不会发生灾难性问题。

Spark Streaming

容错

RDDs可以记住从原始的容错输入创建它的操作序列
批量输入数据被复制到多个工作节点的内存中，因此是容错的
由于工作人员故障而丢失的数据，可以从输入的数据开始重新计算

总结

以上就是大数据实时计算框架的简介以及两个具体的实现的例子，下次就要介绍大图计算框架了，敬请期待~