Storm的架构：

 

 

 

 

 

Nimbus：

风暴Nimbus执行一些功能，包括调度、监视和分发jar。它还提供系统的运行监控组件，并管理多个拓扑的计数器。因此，Nimbus组件在功能上过载，并且由于各种原因常常成为操作瓶颈。

首先，Nimbus调度程序在粒度级别上不支持Storm Worker的资源预留和隔离。因此，在同一台计算机上运行的属于不同拓扑的Storm Worker能会相互干扰。这种情况反过来会导致无法跟踪的性能问题。为了缓解这个问题，我们必须单独不同的拓扑；也就是说，整个机器都专用于一个拓扑，这样使得资源的利用率下降。

Storm负责在不考虑底层调度框架的情况下，确定如何将资源分配给给定拓扑，以及如何跨不同拓扑分配资源。除了可扩展性问题外，这种方法还有几个缺点。首先，还不清楚Storm的调度决策与底层调度框架（如果一个框架是软件栈的一部分）之间的关系。其次，在缺乏调度框架的情况下，在资源共享的环境中，很难跨不同拓扑提供资源隔离。最后，与Heron不同，Heron在部署拓扑时获取资源，Storm集群的资源必须在提交任何拓扑之前获取。因此，系统管理员需要预先知道集群上的多个拓扑将如何运行，以及它们的预期资源需求，以便预先分配资源。

其次，Storm广泛使用Zookeeper来管理Worker和Supervisior的心跳。Zookeeper的使用限制了每个拓扑并发度，当集群中的拓扑数目过多时，Zookeeper容易成为瓶颈。

最后，Nimbus的单点故障。当Nimbus故障时，用户既不能提交任何新的拓扑，也不能杀死现有的拓扑。此外，当Nimbus发生故障时，任何发生故障的现有拓扑都无法自动检测和恢复。

Zookeeper：

用于维持拓扑的状态信息，包括每一个Worker的注册信息和每一个拓扑的物理拓扑结构等和接收来自Supervisior的心跳。

Supervisior：

运行在每一个Worker上，用于与Zookeeper进行通信和接收以及分发tuple。

Worker：

每一个Worker（运行逻辑拓扑的一部分）上可以运行多个Executor（运行逻辑节点），每一个Executor上可以运行多个task（物理拓扑节点），由于每一个Worker运行在一个JVM上，所以对于系统出错位置的定位不准确。例如一个Kafka Spout、一个将处理tuple逻辑的bolt，以及另一个将输出写入键值存储的bolt可能在同一个JVM中运行。在这种情况下，很难对特定任务的行为和性能进行推理，因为不可能隔离其资源使用。而且不同task的日志写在同一个日志文件中使得查看和追踪十分的困难。同时，Storm假设每一个Worker是同构的，这样就会造成大量的资源浪费。例如，考虑安排3个Sput和1个bolt和2个Worker，一个Worker运行一个Spout和一个bolt，另一个运行两个Spout。假设bolt和pout任务分别需要10GB和5GB的内存，则此拓扑需要为每个worker保留总共15GB的内存，因为其中一个worker必须运行bolt和pout任务。这种分配策略导致拓扑总共需要30GB的内存，而实际只需要25GB的内存；因此，浪费了5GB的内存资源。

 

Heron的架构：

 

 

Heron之大观：

Heron的总体架构如图3所示。用户使用Heron创建拓扑，并使用Heron命令行工具将拓扑部署到Aurora调度器。Aurora[是一个通用的服务调度程序，它作为一个框架运行在Mesos之上。然而，我们的架构实现了一个调度器抽象，它有助于在其他调度器上运行Heron，比如YARN、Mesos和ECS（Amazon EC2 Docker容器服务）。在Storm中使用Nimbus进行调度。由于Twitter自主开发的Aurora调度器和其他开源调度器（例如YARN）已经变得复杂，我们有意识地选择与这些调度器一起工作，而不是实现另一个调度器。每个拓扑都作为一个由几个container组成的Aurora作业运行，如图4所示。第一个container运行一个名为Topology Master的进程。其余的容器分别运行一个流管理器（Stream Manager）、一个度量管理器（Metric Manager）和一些称为Heron实例（Heron Instance）的进程（它们是运行用户逻辑代码的Spout和bolt）。可以在单个物理节点上启动多个container。这些container由Aurora根据集群中节点的资源可用性分配和调度。（在Twitter上，Aurora将这些容器映射到Linux cgroup。）可以运行备用拓扑主节点以获得可用性。拓扑的元数据（包括有关启动作业的用户、启动时间和执行详细信息的信息）保存在Zookeeper中。Heron实例是用Java编写的，因为它们需要运行用户逻辑代码（用Java编写）。每个Heron实例有一个JVM。所有Heron进程都使用协议缓冲区（protobufs）相互通信。

Topology Master（TM）：

负责管理整个拓扑。它提供了一个单一的接触点来发现拓扑的状态（因此类似于YARN中的应用程序主节点）。在启动时，TM通过在Zookeeper中的一个众所周知的位置创建一个短暂的节点，使自己成为可发现的。短暂节点有以下两个用途：·它防止多个TMs成为同一拓扑的主节点，从而为拓扑的不同进程提供整个拓扑的一致视图。它允许属于拓扑的任何其他进程发现TM。请注意，由于TM不涉及数据处理路径，因此它不是瓶颈。

Stream Manager（SM）：

SM的关键功能是有效地管理元组的路由。每个Heron实例（HI）都连接到其本地SM以发送和接收元组。一个拓扑中的所有MSg连接在它们之间形成一个O（k^2）连接网络，其中k是拓扑的物理拓扑中的容器/SMs的数量。注意，由于HIs的数目n通常远大于k，因此该设计允许通过在O（k^2）物理连接上多路复用O（n^2）逻辑信道来缩放通信覆盖网络。此外，在同一容器中从一个HI路由到另一个HI的任何元组都使用本地短路机制路由。

Topology Backpressure 

与Storm不同，Heron采用Topology Backpressure 动态调整数据流经拓扑的速率。这种机制在拓扑结构中非常重要，不同的组件可以以不同的速度执行（并且每个组件中的处理速度可以随着时间的推移而改变）。例如，考虑一个拓扑，其中下游节点运行缓慢，或者由于数据或执行倾斜而速度减慢。在这种情况下，如果上游节点不减速，将导致缓冲区建立长队列，或导致系统丢弃元组。如果元组在流中间被丢弃，那么效率可能会损失，因为这些元组已经产生的计算被浪费了。需要一个Topology Backpressure 来减缓早期阶段。

Spout Backpressure：在这种方法中，SMs会限制它们的局部Spout，以减少注入到拓扑中的新数据。此方法与SMs和HIs之间的TCPBackpressure 结合使用。当SM意识到一个或多个HI正在减速时，它会识别出其本地Spout并停止从中读取数据。这个机制的作用是减缓Spout的速度，因为用于向SM发送元组的Spout发送缓冲区将被填满，并最终阻塞。受影响的SM会向其他短信发送一条特殊的信息，要求他们限制自己的局部Spout。当其他短信收到这条特殊信息时，他们也会限制它们的局部Spout。一旦慢HI赶上，本地SM发送停止Backpressure消息到其他SM。当其他SM收到这条特殊信息时，它们会重新开始使用来自本地Spout的数据。

Heron实例：

Spout和bolt的主要工作在Heron实例（HIs）中进行。与Storm worker不同，每个HI都是一个JVM进程，它只运行Spout和bolt的单个task。这样的设计使我们可以轻松地调试/分析Spout和bolt，因为开发人员可以轻松地看到来自HI的事件和日志序列。请注意，由于数据移动的复杂性已经转移到了SMs上，因此我们很容易在将来考虑用其他语言编写本地HIs。为了实现HIs，我们考虑了使用两个线程。

在这个设计中，HIs有两个线程，即网关线程和任务执行线程，如图5所示。网关线程负责控制HI的所有通信和数据进出。它维护到本地SM和度量管理器的TCP连接。它还负责从本地SM接收传入的元组。这些元组被发送到任务执行线程进行处理。任务执行线程运行用户代码。当任务执行线程启动时，它将执行“open”或“prepare”方法，具体取决于实例分别执行的是Spout和bolt。在bolt的情况下，当元组到达时，任务执行线程调用带有传入元组的“execute”方法进行处理。在Spout的情况下，它反复调用“nextTuple”方法从源中获取数据，然后将这些数据作为元组注入到拓扑中。从Spout或bolt发出的元组被发送到网关线程，网关线程将元组转发到本地SM。

除了处理元组之外，任务执行线程还收集各种度量，例如执行的元组数、发出的元组数、确认的元组数以及在处理元组期间所经历的延迟。网关线程和任务执行线程之间使用三个单向队列进行通信，如图5所示。网关线程使用队列中的数据将元组推送到任务执行线程进行处理。任务执行线程使用数据输出队列将元组发送到网关线程（用于发送到拓扑的其他部分）。任务执行线程使用度量输出队列将收集到的度量传递给网关线程。

 

Matric Manager（MM）：

从系统中的所有组件收集和导出metric。这些metric包括拓扑的系统metric和用户metric。每个container都有一个Matric Manager，SM和Heron实例向其报告它们的metric。metric从每个container发送到内部监视系统。MMs还将metric传递给TM，以便在外部UI中显示。使用本地MM分离metric报告为我们将来支持其他监控系统（如Ganglia和Graphite）提供了灵活性。

Heron的总体运行流程：

     系统启动之后，用户将topology上传到资源管理者（如YARN,Mesos），然后YARN根据topology和系统的资源使用状况利用Resource Manager为拓扑定制一个资源分配方案，然后调用scheduler为topology分配资源（container）,然后scheduler启动用于管理topology的topology master，TM将自己的信息（包括逻辑和物理的拓扑）注册到state manager（zookeeper）中， 然后scheduler启动每一个container的stream manager到zookeeper获取自己的TM和物理拓扑并根据这个物理拓扑确定自己tuple的流动方向，最后内部的Heron instance 确定自己的部分物理拓扑。

    SM通过心跳信号来定期与TM进行通信。在SM内部，对于每一个Heron instance，由两个线程组成，一个线程用于从SM取得数据，将经过计算后的数据发送给SM，还要保持与Matric manager的通信，一个线程用于执行逻辑计算，SM，MM,HI之间通过三个队列进行数据传输。同时对于数据队列提供动态变化的机制来防止内存溢出。

    此外，当一个节点的处理速度变得缓慢时，他会通过停止接收来自源节点的数据来实现对源节点处理速度的减缓，此时整个拓扑的处理速度等于最小节点的处理速度。

    当TM出现故障时，备用的TM从zookeeper处获取状态信息然后接管工作，SM通过zookeeper获取新的TM的信息，当TM恢复后，他将作为备用的TM。当SM出现故障时，它在同一个container中重新启动，重新发现TM，并启动一个连接来获取物理拓扑，以检查其状态是否有任何更改。其他已失去与故障的SM的连接的SM也会得到一个指示新SM位置的相同物理拓扑的副本，并创建与新SM的连接。当一个实例（HI）故障时，它将重新启动，并与其本地SM联系。然后，HI获取物理拓扑的副本，标识它是数据源还是处理节点，并开始执行相应的用户逻辑代码。