| 编辑推荐: |
本文来自于博客园，本文首先介绍了Storm的基本概念和数据流模型，然后结合一个典型应用场景来说明Storm支持Topology之间数据流订阅的必要性，最后对比了Storm与另一个流处理系统在数据流模型上的区别之处。
|
|
Storm基本概念
Storm是一个开源的实时计算系统，它提供了一系列的基本元素用于进行计算：Topology、Stream、Spout、Bolt等等。
在Storm中，一个实时应用的计算任务被打包作为Topology发布，这同Hadoop的MapReduce任务相似。但是有一点不同的是：在Hadoop中，MapReduce任务最终会执行完成后结束；而在Storm中，Topology任务一旦提交后永远不会结束，除非你显示去停止任务。
计算任务Topology是由不同的Spouts和Bolts，通过数据流（Stream）连接起来的图。下面是一个Topology的结构示意图：
其中包含有：
Spout：Storm中的消息源，用于为Topology生产消息（数据），一般是从外部数据源（如Message
Queue、RDBMS、NoSQL、Realtime Log）不间断地读取数据并发送给Topology消息（tuple元组）。
Bolt：Storm中的消息处理者，用于为Topology进行消息的处理，Bolt可以执行过滤， 聚合，
查询数据库等操作，而且可以一级一级的进行处理。
最终，Topology会被提交到storm集群中运行；也可以通过命令停止Topology的运行，将Topology占用的计算资源归还给Storm集群。
Storm数据流模型
数据流（Stream）是Storm中对数据进行的抽象，它是时间上无界的tuple元组序列。在Topology中，Spout是Stream的源头，负责为Topology从特定数据源发射Stream；Bolt可以接收任意多个Stream作为输入，然后进行数据的加工处理过程，如果需要，Bolt还可以发射出新的Stream给下级Bolt进行处理。
下面是一个Topology内部Spout和Bolt之间的数据流关系：
Topology中每一个计算组件（Spout和Bolt）都有一个并行执行度，在创建Topology时可以进行指定，Storm会在集群内分配对应并行度个数的线程来同时执行这一组件。
那么，有一个问题：既然对于一个Spout或Bolt，都会有多个task线程来运行，那么如何在两个组件（Spout和Bolt）之间发送tuple元组呢？
Storm提供了若干种数据流分发（Stream Grouping）策略用来解决这一问题。在Topology定义时，需要为每个Bolt指定接收什么样的Stream作为其输入（注：Spout并不需要接收Stream，只会发射Stream）。
目前Storm中提供了以下7种Stream Grouping策略：Shuffle Grouping、Fields
Grouping、All Grouping、Global Grouping、Non Grouping、Direct
Grouping、Local or shuffle grouping，具体策略可以参考这里。
一种Storm不能支持的场景
以上介绍了一些Storm中的基本概念，可以看出，Storm中Stream的概念是Topology内唯一的，只能在Topology内按照“发布-订阅”方式在不同的计算组件（Spout和Bolt）之间进行数据的流动，而Stream在Topology之间是无法流动的。
这一点限制了Storm在一些场景下的应用，下面通过一个简单的实例来说明。
假设现在有一个Topology1的结构如下：通过Spout产生数据流后，依次需要经过Filter
Bolt，Join Bolt，Business1 Bolt。其中，Filter Bolt用于对数据进行过滤，Join
Bolt用于数据流的聚合，Business1 Bolt用于进行一个实际业务的计算逻辑。
目前这个Topology1已经被提交到Storm集群运行，而现在我们又有了新的需求，需要计算一个新的业务逻辑，而这个Topology的特点是和Topology1公用同样的数据源，而且前期的预处理过程完全一样（依次经历Filter
Bolt和Join Bolt），那么这时候Storm怎么来满足这一需求？据个人了解，有以下几种“曲折”的实现方式：
1） 第一种方式：首先kill掉已经在集群中运行的Topology1计算任务，然后实现Business2
Bolt的计算逻辑，并重新打包形成一个新的Topology计算任务jar包后，提交到Storm集群中重新运行，这时候Storm内的整体Topology结构如下：
这种方式的缺点在于：由于要重启Topology，所以如果Spout或Bolt有状态则会丢失掉；同时由于Topology结构发生了变化，因此重新运行Topology前需要对程序的稳定性、正确性进行验证；另外Topology结构的变化也会带来额外的运维开销。
2） 第二种方式：完全开发部署一套新的Topology，其中前面的公共部分的Spout和Bolt可以直接复用，只需要重新开发新的计算逻辑Business2
Bolt来替换原有的Business1 Bolt即可。然后重新提交新的Topology运行。这时候Storm内的整体Topology结构如下：
这种方式的缺点在于：由于两个Topology都会从External Data Source读取同一份数据，无疑增加了External
Data Source的负载压力；而且会导致同样的数据在Storm集群内被传输相同的两份，被同样的计算单元Bolt进行处理，浪费了Storm的计算资源和网络传输带宽。假设现在不止有两个这样的Topology计算任务，而是有N个，那么对Storm的计算Slot的浪费很严重。
注意：上述两种方式还有一个公共的缺点——系统可扩展性不好，这意味着不管哪种方式，只要以后有这种新增业务逻辑的需求，都需要进行复杂的人工操作或线性的资源浪费现象。
3） 第三种方式：OK，看了以上两种方式后，也许你会提出下面的解决方案：通过Kafka这样的消息中间件，实现不同Topology的Spout共享数据源，而且这样可以做到消息可靠传输、消息rewind回传等，好处是对于Storm来说，已经有了storm-kafka插件的支持。这时候Storm内的整体Topology结构如下：
这种实现方式可以通过引入一层消息中间件减少对External Data Source的重复访问的压力，而且可以通过消息中间件层，屏蔽掉External
Data Source的细节，如果需要扩展新的业务逻辑，只需要重新部署运行新的Topology，应该说是现有Storm版本下很好的实现方式了。不过消息中间件的引入，无疑将给系统带来了一定的复杂性，这对于Storm上的应用开发来说提高了门槛。
值得注意的是，方案三中仍遗留有一点问题没有解决：对于Storm集群来说，这种方式还是没有能够从根本上避免数据在Storm不同Topology内的重复发送与处理。这是由于Storm的数据流模型上的限制所导致的，如果Storm实现了不同Topology之间Stream的共享，那么这一问题也就迎刃而解了。
一个流处理系统的数据流模型
个人工作中有幸参与过一个流处理框架的开发与应用。下面我们来简单看看其中所采用的数据流模型。
首先，先来看一下该流处理系统内的几个基本概念：
1）数据流（data stream）：时间分布和数量上无限的一系列数据记录的集合体；
2）数据记录（data record）：数据流的最小组成单元，每条数据记录包括 3 类数据：所属数据流名称（stream
name）、用于路由的数据（keys）和具体数据处理逻辑所需的数据（value）；
3）数据处理任务定义（task definition）：定义一个数据处理任务的基本属性，无法直接被执行，必须特化为具体的任务实例。其基本属性包括：
（可选）输入流（input stream）：描述该任务依赖哪些数据流作为输入，是一个数据流名称列表；数据流产生源不会依赖其他数据流，可忽略该配置；
数据处理逻辑（process logic）：描述该任务具体的处理逻辑，例如由独立进程进行的外部处理逻辑；
（可选）输出流（output stream）：描述该任务产生哪个数据流，是一个数据流名称；数据流处理链末级任务不会产生新的数据流，可忽略该配置；
4）数据处理任务实例（task instance）：对一个数据处理任务定义进行具体约束后，可推送到某个处理结点上运行的逻辑实体。附加下列属性：
数据处理任务定义：指向该任务实例对应的数据处理任务定义实体；
输入流过滤条件（input filting condition）：一个 boolean 表达式列表，描述每个输入流中符合什么条件的数据记录可以作为有效数据交给处理逻辑；若某个输入流中所有数据记录都是有效数据，则可直接用
true 表示；
（可选）强制输出周期（output interval）：描述以什么频率强制该任务实例产生输出流记录，可以用输入流记录个数或间隔时间作为周期；忽略该配置时，输出流记录产生周期完全由处理逻辑自身决定，不受框架约束；
5）数据处理结点（node）：可容纳多个数据处理任务实例运行的实体机器，每个数据处理结点的IPv4地址必须保证唯一。
该流处理系统，采用分布式策略，由多个数据处理结点进行数据的处理过程；将流式数据的处理过程划分为不同的阶段，每个阶段伴随数据流的流入、任务的处理及数据流的流出；各个阶段会有若干个处理结点参与完成，其中，每个处理结点上会有若干个数据处理任务实例运行，每个数据处理任务实例则是对一个数据处理任务定义进行具体约束后，可推送到某个处理结点上运行的逻辑实体。在不同的处理结点之间，数据流根据配置信息进行传输；在处理结点内部，结点根据配置信息对流经该结点的数据进行处理。
下图为系统对于流式数据的基本处理流程：
1）定义数据流：将流式数据的处理过程划分成不同的阶段，定义出不同的数据流名称；
2）定义数据处理任务：为数据流的处理过程定义相应的数据处理任务，其中，各个处理任务定义了外部处理逻辑，且其输入/输出数据流须从1）中预定义的数据流列表中选取；
3）定义数据处理结点：定义各个数据处理结点的名称及其IPv4地址信息；
4）定义数据处理任务实例：为3）中定义好的每个处理结点，分别定义运行在其上的数据处理任务实例，其中，每个任务实例所对应的数据处理任务实体须从2）中预定义的处理任务列表中选取；
5）加载数据流的相关配置信息及订阅信息（具体格式见“附录：配置信息格式”），然后开始从数据流产生源读取数据；
6）运行数据流源结点上的任务实例：数据流处理链源结点上的处理任务实例直接对数据流产生源的数据进行处理，然后产生新的输出数据流；
7）运行下一级结点上的任务实例：中间的处理结点上的处理任务实例依赖于上一级处理结点的输出数据流作为输入数据流，从中读取数据，进行处理，产生输出数据流，并传递到下一级处理结点；
8）判断是否到达数据流末级结点：
如果不是，则继续返回步骤7），按照数据流的流动关系，继续运行下一级结点上的任务实例；
否则为数据流末级结点，则进行步骤9）。
9）输出结果：数据流处理链末级结点上的处理任务不会产生新的数据流，完成最终的数据处理任务后将结果进行输出。
以上是数据流在不同处理结点之间的处理流程。每个处理结点作为流处理的一个环节，其结点内部的处理流程如下图所示：
1）每个处理结点启动流处理过程后，开启网络服务，监听从上一级处理结点发出的TCP连接请求，接收从上一级结点发来的数据；
2）处理结点不间断地接收从上一级处理结点发来的数据，对于每条数据记录，根据数据流名进行筛选，将其分发到该数据流所对应的处理进程中；
3）将从每个特定数据流发来的数据，广播到所有以该数据流作为输入数据流的数据处理任务实例中；
4）数据处理任务实例从其输入数据流中接收数据，按照过滤条件进行筛选，然后将符合过滤条件的数据记录发送给外部应用程序进行处理；
5）外部应用程序启动外部处理进程，对数据进行实际处理过程，并将每条数据记录的处理结果返回给相应的数据处理任务实例；
6）数据处理任务实例从外部应用程序收集处理后的结果数据，并依次将其转发到对应的输出数据流中；
7）输出数据流进程接收发向该数据流的数据，然后按照数据流的订阅关系，将数据发送到所有订阅了该数据流的下一级处理结点；
8）根据下一级处理结点的IP地址和端口号，通过TCP请求与下一级处理结点建立网络连接，然后将数据按序传输到下一级处理结点。
二者在数据流模型上的不同之处
至于两个系统的实现细节，我们先不去做具体比较，下面仅列出二者在数据流模型上的一些不同之处（这里并不是为了全面对比二者的不同之处，只是列出其中的关键部分）：
1） 在Storm中，数据流Stream是在Topology内进行定义，并在Topology内进行传输的；而在上面提到的流处理系统中，数据流Stream是在整个系统内全局唯一的，可以在整个集群内被订阅。
2） 在Storm中，数据流Stream的发布和订阅都是静态的，所谓静态是指数据流的发布与订阅关系在向Storm集群提交Topology计算任务时，被一次性生成的，这一关系在Topology的运行过程中是不能被改变的；而在上面提到的流处理系统中，数据流Stream的发布和订阅都是动态的，即数据处理任务task可以动态的发布Stream，也可以动态的订阅系统内已经生成的任意Stream，数据流的订阅关于通过分布式应用程序协调服务ZooKeeper集群的动态节点来维护管理。
好了，有了以上的对比，我们不难发现，对于本文所举的应用场景实例，Storm的数据流模式尚不能很方便的支持，而在这里提到的这个流处理系统的全局数据流模型下，这一应用场景的需求可以很方便的满足。
总结的话
个人觉得，Storm有必要实现不同Topology之间Stream的共享，这个至少可以在不损失Storm现有功能的前提下，使得Storm在处理实际生产环境下的一些应用场景时更加从容应对。
至于如何在现有Storm的基础上实现这一需求，可能的方式很多。一种简单的方式是通过Zookeeper来集中存储、动态感知Topology之间Stream的“发布-订阅”关系，同时在Storm的消息分发过程中对这种情况加以处理。
|
