Apache Storm学习笔记

Apache Storm是一个分布式的实时计算系统，它能够可靠的对无边界的数据流进行处理。与Hadoop的批量处理方式不同，Storm对数据进行的是实时处理。Storm很简单，支持很多编程语言。

Storm的应用场景包括：实时分析、在线机器学习、持续计算、分布式RPC、分布式ETL。

Storm的性能很好，每个节点在一秒内能够处理多达100万个元组。Storm支持无限制扩容、容错、并且保证数据能够被处理。

你可以让既有的数据库、消息队列系统和Storm进行集成。Storm拓扑消费数据流，以任意复杂的方式处理数据流，数据流可以在任何计算阶段（Stage）重新分区。

实时计算的逻辑被封装到Storm拓扑中，拓扑和MapReduce的Job类似。一个关键的区别是，MapReduce Job最终会结束，而拓扑则一直运行下去。

你可以使用TopologyBuilder构建一个拓扑。

Storm把每个具体的工作委托给多种类型的组件处理。这些组件都装配在拓扑中。

拓扑可以跨越Storm集群的多个工作节点运行。

流是Storm的核心抽象，它是无边界的元组的序列。流以分布式的风格被并行的创建、处理。流的元素是元组，元组可以包含整数、字符串、浮点数、布尔、比特数组等类型的字段。你可以自己实现串行化器，支持任何自定义数据类型。

你可以使用OutputFieldsDeclarer来声明流以及流的Schema。在声明的时候，每个流被赋予一个标识符。仅仅释放单个流的Spout很常见，因此OutputFieldsDeclarer提供了声明单个流的便捷方法，不需要指定标识符，流的标识符自动设置为default。

Storm中被处理的数据的基本单元，本质上就是一个预定义的字段的值列表。

Spout可以释放出元组，不同的Bolt可以读取同一Spout产生的元组，并释放不同格式的元组，供其它Bolt消费。

在拓扑中，Spout是流的来源。一般来说，Spout从外部读取元组，并释放（Emit）出元组的流（Spout本意即喷射口）。Spout可以是可靠的（Reliable ）或者不可靠的。可靠的Spout支持在处理失败后重复（Replay）元组。所有Spout必须实现IRichSpout接口。

Spout可以释放不止一个流，你可以多次调用 OutputFieldsDeclarer的 declareStream。你可以指定 SpoutOutputCollector.emit调用释放哪个流。

Spout的主要方法是nextTuple，调用它可以释放一个新的元组到拓扑中，如果没有可用的元组，该方法应该简单的返回。Storm强制要求所有Spout实现类的nextTuple方法是非阻塞性的，因为Storm在单个线程中调用Spout的所有方法。

其它重要的方法包括ack、fail，这些方法仅仅针对可靠Spout调用，分别在元组顺利通过拓扑后、处理失败后调用，将处理结果通知给Spout。

Spout将数据传递给Bolt这种组件进行实质上的逻辑处理。拓扑中最主要的内容就是Bolt构成的链条。

拓扑中所有处理逻辑发生在Bolt中，Bolt可以进行过滤、聚合、Join等操作，可以和数据库进行交互。

Bolt可以进行简单的流变换（Stream Transformation），复杂的流变换通常需要多个步骤，也就需要多个Bolt。

Bolt可以释放不止一个流，你可以多次调用 OutputFieldsDeclarer的 declareStream。你可以指定 OutputCollector.emit调用释放哪个流。

在声明Bolt的输入流时，你需要订阅其它组件的某个特定流，如果要订阅某个组件的多个流，你需要逐个订阅。InputDeclarer提供了订阅具有默认标识符的流的快捷方法，例如 declarer.shuffleGrouping("1")订阅组件1的 DEFAULT_STREAM_ID流。在构建拓扑时，你也可以声明这种流订阅。

Bolt的主要方法是execute，此方法接受一个元组输入。Bolt使用OutputCollector对象释放新的元组。Bolt会针对每个输入元组调用OutputCollector.ack方法，以通知Storm目标元组已经处理成功，这些ack调用最终导致Spout.ack被调用。大部分Bolt根据输入元组输出0-N个输出元组。IBasicBolt提供自动确认（Ack）功能。

在Bolt中启动新线程进行异步处理是很好的做法，OutputCollector是线程安全的，你可以在任何时候调用它。

Bolt的基础接口是IRichBolt。要设计进行过滤、简单function的Bolt可以选择实现IBasicBolt接口。Bolt调用OutputCollector来释放元组到它的输出流中。

定义拓扑的一部分工作是，指定每个Bolt能够接受哪些流作为其输入。流分组（Stream Grouping）指定了如何在Bolt的Task之间进行流的分区（Partition）。

Storm可以保证每个元组都被拓扑处理。为了确认元组的处理过程，Storm对Spout释放出的元组树进行跟踪，并以此确定何时元组树被成功处理。每个拓扑都具有一个消息超时属性，如果在此超时到达之前没有检测到元组已经处理成功，则认定处理失败，并Replay目标元组。

之所有叫元组树，是因为在任何一个Bolt中，一个源元组可能产生多个目标元组，这样经过多个Bolt后，就形成树状依赖关系。注意，目的和源的依赖关系需要编程式的声明，一个目标元组可以依赖于多个源元组。

每个Spout、Bolt在集群中执行一定数量的Task，每个Task对应一个执行线程。流分组决定了元组如何从一组Task分发到另外一组Task。调用TopologyBuilder的setSpout/setBolt方法可以设置并发度。

Topology跨越1-N个工作进程执行，每个工作进程都是一个JVM实例，负责执行拓扑中所有Task的一个子集。Storm会尝试尽可能平均的分配Task给Worker。例如，一个总和并行度为300的拓扑，分配了50个工作进程，则每个进程需要执行6个Task。

Storm集群包含两类节点：

1. 主节点（Master）：这类节点运行一个名为Nimbus的守护程序。此程序负责在集群中分发代码（处理逻辑）、向工作节点分配Task、监控Task是否执行失败
2. 工作节点（Worker）：这类节点运行一个名为Supervisor的守护程序。此程序负责执行拓扑的某个部分

Storm支持在本地磁盘或者ZooKeeper上保存集群的所有状态信息，因此守护程序宕机不会对系统的健康状况产生影响。

Storm和Hadoop的核心概念的对应关系如下图：

	Storm	Hadoop
组件	JobTracker	Nimbus
	TaskTracker	Supervisor
	Child	Worker
应用	Job	Topology
接口	Mapper/Reducer	Spout /Bolt

Storm提供四类内置的调度器： DefaultScheduler, IsolationScheduler, MultitenantScheduler, ResourceAwareScheduler。

Storm有两种运行模式（Operation Modes）。

在此模式下，Storm的拓扑在本地机器中单个JVM中运行。此模式主要用于开发、测试和调试。

在此模式下，我们需要将定义好的拓扑提交到Storm集群中。集群通常由运行在很多机器中的进程组成。远程模式下不会显示调试信息，通常用于产品环境。

你也可以在单台机器上进行远程模式的部署。

Storm包含若干不同的守护进程，Nimbus能够调度Worker进程，Supervisors负责启动、杀死Worker进程。Logger View让你能够访问日志。Web服务器让你能够通过GUI查看Storm状态。

当Worker进程死掉后，Supervisor会重新启动它。如果Worker反复死掉并且不能完成到Nimbus的心跳检测，Nimbus会重新调度此Worker。

当节点宕机后，Nimbus会重新分配在节点上执行的Task给其它节点。

Nimbus/Supervisor被设计为无状态的，所有状态信息存放在ZooKeeper或者磁盘上。因此它们宕机后，你可以安全的重启它们。你应该考虑使用daemontools/monit之类的工具监控这类进程的状态，并在它们死掉后立即重启。

如果Nimbus节点（Master）宕机，Worker节点仍然会继续工作。此外Supervisor仍然会自动重启死掉的Worker进程。但是没有Nimbus就不能把Worker节点分配到其它机器上。从1.0.0开始，Storm支持Nimbus的HA。

在本章，我们创建一个简单的，分析动力环境监测值的拓扑。

此类型模拟监测值采集程序，以随机的方式产生监测值：

MessageReader是此拓扑中唯一的Spout，从MessageSource中读取监测值报文并直接传递给Bolt处理。

任何Spout需要实现IRichSpout接口，你可以选择继承缺省适配BaseRichSpout。

这是一个Bolt，读取监测值报文并解析之，释放多个监测值信息元组。 

任何Bolt都需要实现IRichBolt接口。你可以选择实现IBasicBolt（继承BaseBasicBolt），这样Storm会自动进行元组确认（Ack），使用IBasicBolt时你可以抛出FailedException表示当前元组处理失败。

这是一个Bolt，它读取MessageParser释放的监测值元组，然后进行实时统计分析。

主程序，创建拓扑并提交到集群中执行：

Storm对于下述运行在集群中的拓扑中的实体：

1. 工作进程（Worker processes）
2. 执行器（Executors，线程）
3. 任务（Task）

进行了明确的区分：

1. Storm集群中的节点（机器）可以运行0-N个工作进程，这些工作进程可以属于1-N个拓扑。每个工作进程就是一个JVM，它仅仅为单个拓扑运行Executor。也就是说Worker不能被两个拓扑共享
2. 单个Worker进程中可以运行1-N个Executor，每个Executor都是Worker进程产生的线程。每个Executor运行单个拓扑的单个组件的1-N个Task。也就是说Executor不能被两个组件共享
3. 任务进行实际的数据处理，它执行Spout或者Bolt中的代码逻辑。它由Executor来执行

可以看到：

1. 工作进程执行一个拓扑的子集，它只能属于一个拓扑
2. 工作进程可以运行多个执行器线程，这些线程可以运行不同的组件。但是每个线程只能运行一种组件的任务
3. 任务必须在执行器线程内部运行

要对并行度进行配置，参考并发度配置。

流分组指定了如何在Bolt的Task之间进行流的分区（Partition）——流的每个元组应该由哪个Task处理。

在定义拓扑的时候，你可以为Bolt指定流分组，指定流分组的同时也就指定了Bolt的数据源。你可以针对一个bolt多次指定流分组，也就是指定多个输入流：

要实现自己的流分组，需要实现CustomStreamGrouping接口。通常情况下，你应该优先考虑使用Storm内置的流分组。

流分组	说明
Shuffle	最常用的一种流分组。元组被随机的分发给Bolt的Task，此实现保证每个Task接受相等数量的元组
Fields	根据元组的字段值进行分组，字段值相同的元组被分发给同一个Task 可以基于1-N个字段的值进行分组
Partial Key	元组首先按照字段分组的方式被分区，并进一步的被两个下游的Bolt进行负载均衡
All	输入流的每个元组被复制到Bolt的所有实例（Task） 需要执行某种广播逻辑时，考虑使用这种流分组
Global	整个流被转发给单个具有最小Id的那个Task
None	表示不关心如何进行分组，当前此分组的行为和Shuffle类似。Storm在处理这种方式分组的流时，上游Spout/Bolt和下游Bolt使用同一执行线程
Direct	上游Spout/Bolt（生产者）直接决定元组分发给下游Bolt（消费者）的哪个Task 这种分组方式仅仅能应用到被声明为直接流（Direct Stream）的流上。要释放元组到直接流上，必须调用OutputCollector.emitDirect方法。获得消费者的Task Id途径可以是： 通过TopologyContext获得 根据OutputCollector.emit方法的返回值获得，返回值是接受元组的Task 的Id 示例代码： Java 1 2 3 4 // 获得某个下游组件的所有Task Id Set<Integer> taskIds = context.getTaskToComponent( "downstream-bolt-id" ).keySet(); // 发送给目标Task collector.emitDirect( taskId,tuple );
Local or shuffle	如果下游Bolt在相同（与释放流的组件）工作进程（Work Process）中行运行着一个或者更多的Task，则元组被分发给这些进程内的Task。否则，行为与Shuffle grouping相同

实现CustomStreamGrouping、Serializable接口：

使用上述自定义流分组：

此类用于在当前JVM中建立一个本地模式的集群。通常在开发阶段使用，可以方便的对各种拓扑进行调试：

此类用于将拓扑提交到一个远程Storm集群中运行。与本地模式不同，你不能对远程集群进行控制。

下文有一个实际的例子。

通过storm客户端命令，也可以将拓扑提交到远程集群。首先你需要打JAR包，然后执行：

注意，JAR的入口函数必须调用submitTopology。

本节列出常见的拓扑模式（Pattern）。

很多情况下，出于效率的考虑，你可能希望一批批的处理元组，而不是一个个的。例如，你可能希望批量的将元组入库，以利用数据库的吞吐能力。

你可以简单的将元组存储到实例变量中，等到积累的足够多了，批量的处理它们然后统一Ack。

如果执行批处理的Bolt释放元组，你可能需要使用多重锚定，确定输出元组和输入元组之间的因果关系。

很多Bolt都是读取一个输入元组、处理后根据此元组释放0-N个元组，最后对输入元组进行Ack。基于IBasicBolt可以获得此模式的封装。

经过字段分组之后，该字段一样的元组总是由同一个Bolt处理。假设Bolt需要根据该字段进行某种转换，那么在Bolt实例变量中缓存这种转换结果，其缓存性能会较高。

例如，一个Bolt负责接收短URL并展开它，展开需要通过HTTP来调用第三方服务，因此缓存展开URL很有必要。可以在Bolt中创建一个短URL到展开URL的HashMap作为缓存。配合字段分组，让相同短URL总是分发给同一Bolt实例，这样上述缓存的命中率会比不分组高得多。

某些情况下，你需要处理很多输入元组，并且释放出其中最xx的10个元组。

最简单的方案是，使用全局流分组，也就是让所有元组都输入到单个Bolt实例中。这种方案可能会导致性能瓶颈。

一个改进的方案是，让多个Bolt都进行TopN计算，在一个下游的Bolt中，重新进行一次TopN计算：

我们知道，基于ack/fail可以保证消息：

1. 要么完全通过拓扑
2. 要么，可选的，进行Replay

问题是，如果进行Replay时，如何保证事务性？或者说，如何保证某些逻辑不会重复执行。

从0.7开始，Storm引入了事务性拓扑，可以保证Replay安全的进行，确保它们仅仅被处理一次。

在事务性拓扑中，Storm混合使用并行、串行的元组处理流。Spout生成的一批元组，被并行的发送给Bolt处理。某些Bolt被称为Committer，它们按照严格有序的方式提交已经被处理过的元组批次。

例如，Spout释放两个元组批次，每个批次包含5个元组，这两个元组被并发的交付Bolt处理，但是，如果链条中存在Committer Bolt，则在此Bolt批次必须严格有序的串行化 —— 它不会在第一个批次成功提交之前，提交第二个批次。

Storm事务由两个部分阶段组成：

1. 处理阶段：完全并行阶段，很多批次可以被同时的执行
2. 提交阶段：完全有序阶段，保证批次按照顺序逐一执行

我们以一个Twitter分析工具的例子来了解事务性拓扑的工作机制。此工具的逻辑如下：

1. 读取Tweets并存储到Redis
2. 通过若干Bolt处理Tweets
3. 将处理结果存放到另一个Redis数据库，处理结果包括：所有主题标签（Hashtag）的及其出现次数、用户及其出现在Tweet中的次数、主题标签和用户出现在同一Tweet中的次数

用于构建此工具的拓扑示意如下：

说明如下：

1. TweetsTransactionalSpout连接到数据源，读取并释放元组批次到拓扑
2. 随后的两个Bolt会接受所有元组
   1.  UserSplitterBolt，从Tweet中搜索用户（@之后的单词）
   2. HashatagSplitter，从Tweet中搜索主题标签（#之后的单词）
3. 后面的UserHashtagJoinBolt，对前面两个Bolt的结果进行Join，获得每个用户+标签的组合同时出现的次数。此Bolt是BaseBatchBolt的子类型
4. 最后的RedisCommitterBolt是一个Committer，它接受3个流。进行各种统计，并在处理完一批次后进行入库

相关说明已经附加在源码注释：

事务性拓扑提供了一种可靠的批量处理语义，其关键在于：

1. 提供了组装元组批次，也就是事务的机制，并且信息持久化在ZooKeeper中，不会因为宕机而丢失
2. 提供了串行有序执行语义，你只需要记录一个事务标识符 —— 上一个事务的标识符，因为任何时刻只会有一个事务正在准备提交

流是连续不断的元组序列。任何Spout、Bolt都可以释放0-N个流。

通过 JoinBolt，Storm支持将多个流合并为一个。JoinBolt是一个窗口化（Windowed）的Bolt，它会等待进行合并的多个流的Window duration匹配，确保流依据窗口边界对齐。

每个参与合并的流，必须是根据Join字段进行分fieldsGrouping，也就是说，Join字段一样的元组必须由同一Task处理。

考虑如下SQL语句：

对应的流连接如下：

在Join某个流之前，你必须先引入它： 

实际Join发生的顺序，就是用户声明的顺序。

上面流连接的例子中，都使用了组件名称，而不是流名称。Storm支持基于流名称连接，但是很多组件仅仅释放一个流，且流名称默认为default，要基于流名称连接，必须先定义好命名流。

1. 目前仅仅支持内连接（INNER）和左连接（LEFT）
2. SQL支持针对一个表，通过多个字段分别Join。Storm不支持，每个流有且仅有一个字段能用于Join，该字段用于流分组，确保键相同的分组发送给同一Task
3. 要使用多字段连接，你需要将它们合并为单个字段
4. 连接操作可能占用很高的CPU和内存。当前窗口中累积的数据越多，则连接消耗的时间越长。使用滑动窗口时，越短的滑动间隔导致越频繁的连接操作。因此太长的窗口、太短的滑动间隔都可能导致性能问题
5. 使用滑动窗口时，多个窗口之间可能存在重复的Joined记录，这是因为这些记录的源元组可能跨越多个窗口存在
6. 如果使用消息超时，应当正确设置topology.message.timeout.sec，确保它能够匹配窗口大小，同时考虑拓扑中其它组件可能消耗的时间
7. 连接两个流时，假设窗口大小分别为M、N，那么最坏的情况下可能产生M x N个输出元组，下游Bolt将会释放更多的ACK。这种情况下，Storm的消息子系统可能面临很大的压力，不小心使用可能导致拓扑运行缓慢。下面是管理消息子系统的一些建议：
   1. 增加Worker的堆大小： topology.worker.max.heap.size.mb
   2. 如果拓扑不需要ACK，可以禁用它： topology.acker.executors=0
   3. 禁用事件记录器： topology.eventlogger.executor=0
   4. 禁用拓扑调试功能： topology.debug=false
   5. topology.max.spout.pending设置要匹配完整窗口需要的大小，附加额外的一些空间。这可以避免消息子系统过载时Spout仍然不停释放元组，该项设置为null时更容易发生此情况
   6. 在能够满足需求的情况下，窗口应该尽量的小

本章主要讨论设计拓扑入口点 —— Spout的通用策略，以及如何让Spout支持容错。

在设计拓扑时，考虑消息可靠性需求是一个重要事项。当一个消息无法完成处理时，你需要决定针对此消息进行怎样的处理？整个拓扑有应该有怎样的行为。打个比方，在处理银行存款业务时，每个交易消息都不能丢失；单是在处理百万级数据的统计信息时，丢失一部分消息则不会对结果精确性产生太大影响。

在Storm中，拓扑的设计者负责保证消息的可靠性。这常常需要权衡考虑，因为拓扑需要更多的资源以管理消息不丢失。

你可以在释放元组时为其指定标识符：

如果不指定标识符，Storm会自动生成一个。在消费者中可以调用 input.getMessageId()获得元组的Id。

当消费者针对元组进行ack/fail操作时，生产者的ack/fail回调会被调用，并传入确认/失败的那个元组。至于如何处理失败的元组，需要根据具体业务场景决定，常见的做法是放回消息队列。

引起元组失败的原因有两种：

1. 消费者调用 collector.fail(tuple)
2. 元组处理的时间超过配置的时限， Config.TOPOLOGY_MESSAGE_TIMEOUT_SECS

最简单的接入方式是Spout直接连接到数据源（Message Emitter）。如果数据源是明确的（Well-known）的设备（在这里是一个抽象概念，理解为一般性的数据源即可）或者设备组，这种方式实现起来很简单。

所谓明确的，是指在拓扑启动时设备就就是已知的（Known）并且在拓扑运行期间它保持不变。所谓未知设备，就是在拓扑运行之后才添加进来的。

所谓明确设备组，就是其中所有设备都是已知的一组设备。

下面是一个基于Twitter流API的拓扑，它使用直接连接的接入方式：

注意上面的例子中用到Storm的一个重要特性 —— 在任何组件中访问TopologyContext。利用TopologyContext你可以感知到Spout的Task数量，然后将不同数据源映射到不同的Task，增加拓扑的并行度。

上面是连接到明确的设备的例子，通过类似的方式，可以直接连接到未知设备。但是你需要一个协调（Coordinator）系统来维护可用设备的列表。例如，当进行Web服务器日志分析时，Web服务器的列表可能动态变化。当一个新的Web服务器加入后，Coordinator检测到并为其创建一个Spout。

当建立连接时，应当总是由Spout发起，而不是由数据源发起。因为运行Spout的服务器宕机后Storm可能在另外一台机器上重启此Spout，数据源难以精确的定位到Spout。

除了直接连接，你还可以使用消息队列系统。Spout连接到消息队列读取消息，数据源则把消息发布到消息队列。

消息中间件软件通常都提供可靠性、持久化等保证，这简化了开发任务。使用消息中间件后Spout对数据源毫无感知。

使用消息列队这种接入方式时，如果想增加Spout并行度，可以：

1. 让Spout循环轮询（Round-robin Poll）同一队列
2. 基于哈希再分发：根据哈希值将消息分发给不同Spout，或者分发到不同的子队列。Spout读取子队列。Kafka是一个很好的支持子队列（分区）的消息中间件

Bolt是拓扑中的关键组件，大部分逻辑都发生在Bolt链中。

Bolt这类组件，以元组为输入，可选的，它也能够输出元组。

在创建Bolt时，你需要实现IRichBolt接口。Bolt在客户端创建，然后串行化到拓扑，最后整个拓扑被提交到Storm集群的Master节点。集群会启动工作节点（JVM进程）来运行拓扑组件，Bolt会被反串行化，其prepare方法会被调用，然后开始接受元组。

Storm确保Spout释放的所有元组都能传递给相应的Bolt，至于Bolt是否保证信息堡丢失，由开发人员决定。

拓扑是组件构成的网络，Spout释放的元组，是网络中传递的信息。信息经过不同组件之后，可能分裂、合并，这和Bolt灵活的Tuple处理机制有关：

1. 可以根据一个输入元组产生多个输出元组
2. 可以根据多个输入元组产生一个输出元组

从框架角度来说，无法自动推导输入、输出元组的对应关系。因此在Storm中你需要利用锚定（Anchoring）技术手工指定。

以从Spout发出的原初元组为根，其衍生的任何元组被fail，则此原初元组也就fail，发起此元组的Spout的fail回调自动被调用。

每个Spout/Bolt都可以声明多个输出流，标识符为default的默认流自动声明。要声明新的流，可以调用：

要向指定的流释放元组，调用：

前面提到过，要追踪信息流动方向，需要使用锚定技术。也就是指定输入、输出元组之间的关联关系：

有了这种调用后，Storm才可以对原初元组进行有效的追踪。

如果输出元组基于多个输入元组推导出，可以一起锚定它们：

如果outputTuple失败，则产生inputTuple1、inputTuple2的Spout都会得到通知。 

Storm会在此接口的实现的execute()执行后，自动调用ack，如果execute()抛出异常，则自动调用fail。你可以考虑继承BaseBasicBolt。

Storm提供了一个抽象，允许Bolt存储/取回其操作的状态信息。该抽象有两个实现，一个是基于内存的默认实现，另一个以Redis作为后备存储。

需要状态管理功能的Bolt，需要实现IStatefulBolt接口或者继承BaseStatefulBolt，并实现initState方法。Storm在初始化Bolt实例时，调用initState并传入上次存储的状态，调用发生在prepare之后，处理任何元组之前。

当前仅仅支持的状态实现是org.apache.storm.state.KeyValueState。

分布式远程过程调用（Distributed Remote Procedure Call）利用Storm的计算能力进行远程调用。Storm提供了一系列DRPC的支持工具。

说明：

1. 客户端向DRPC服务器发起同步调用
2. DRPC将请求转发给DRPC Spout，进而在拓扑异步的处理
3. 处理结果由最后一个Bolt返回给DRPC服务器。并返回给客户端 

DRPC服务器是客户端和Storm拓扑之间的桥梁，它作为拓扑中Spout的源运行。

DRPC服务器接受需要执行的函数及其参数，它可以执行多个函数，函数通过名称唯一识别。

对于每一次函数调用，服务器分配一个请求Id，用于在拓扑中唯一的识别调用请求。当拓扑中最后一个Bolt执行完毕后，此Bolt必须释放出包含请求Id、调用结果的数组，DRPC会把结果分发给正确的客户端。

注意：目前此类已经被废弃，其功能由Trident代替。

用于构建DRPC拓扑的抽象。此Builder生成的拓扑会：

1. 自动创建DRPCSpouts，连接DRPC服务器读取请求，并释放元组到拓扑的其它部分
2. 对Bolts进行包装，让调用结果从最后一个Bolt返回

所有添加到LinearDRPCTopologyBuilder的Bolt均按序依次执行

要提交拓扑到Storm集群，可以调用LinearDRPCTopologyBuilder的createRemoteTopology（而不是createLocalTopology），此方法会利用Storm配置中的DRPC相关配置。

要连接到DRPC服务器，使用此类，此类是ThriftClient的子类型。

DRPC服务器暴露了基于Thrift的API，可以在很多语言中使用此API。并且，不管DRPC服务器是本地还是远程的，API完全一样。

下面是一个简单的加法计算器：

Storm使用调度器在集群中进行拓扑的调度，它提供了四种内置的调度器实现：DefaultScheduler, IsolationScheduler, MultitenantScheduler, ResourceAwareScheduler。

要实现自己的调度器，实现IScheduler接口：

要切换调度器，修改配置文件 storm.yaml的 storm.scheduler配置项。

此调度器可以让多个拓扑安全、简单的共享单个集群。此调度器允许你隔离某些拓扑 —— 让它在一组专用的机器上执行，其它拓扑不能使用这些专用机器。被隔离的拓扑优先享用集群资源，被所有隔离拓扑占用内，剩下的机器，所有非隔离拓扑共享之。

要启用此调度器，在Nimbus节点的配置文件中进行如下设置：

Storm支持对一组落到同一“窗口”的元组进行统一的处理。所谓窗口，是被特定条件限定的、连续的元组集合。限定条件有两种：

1. 窗口的长度：包含元组的数量，或者持续的时间
2. 窗口的滑动：每隔多长窗口向前滑动

长度/间隔有两种度量方式：时间、元组数量。

窗口具有一定的长度，并且每隔一段时间，窗口就向前滑动一次。例如下面的滑动窗口示意：

........| e1 e2 | e3 e4 e5 e6 | e7 e8 e9 |...
-5      0       5             10         15    -> time
|<------- w1 -->|
        |<---------- w2 ----->|
                |<-------------- w3 ---->|

窗口的长度是10秒，滑动间隔是5秒。注意元组可能属于多个窗口，每个窗口包含的元组数量可以变化。

窗口具有固定的长度，不同元组仅仅会属于单个窗口。下面是滚动窗口的示意：

| e1 e2 | e3 e4 e5 e6 | e7 e8 e9 |...
0       5             10         15   -> time
    w1        w2           w3

窗口长度是5秒，窗口之间不存在重叠。

要获得窗口化支持，Bolt需要实现IWindowedBolt接口：

通常情况下，你可以选择继承BaseWindowedBolt类，该类提供了一些指定窗口长度、滑动间隔的方法：

默认的，窗口以Bolt处理元组的时间为计算窗口边界的基准。Storm还支持以元组字段给出的时间戳为基准：

除了指定时间戳获取方式之外，Storm还允许指定一个最大的延迟参数：

如果Bolt接收到一个时间戳为06:00:05的元组，且最大延迟为5s，那么后续不应该再接收到时间戳小于06:00:00的元组。如果的确接收到这样的过期（Late）元组，Storm默认不会处理，仅仅在Worker日志中以INFO级别记录。

要改变上述默认行为，可以调用：

当基于元组字段时间戳计算窗口边界时，Storm内部基于接收到的元组的时间戳来计算水位。所谓水位，就是所有输入流中，所有元组中最晚的那个时间戳减去乱序延迟后得到的数值。

Storm周期性的释放出水位时间戳，此时间戳将作为判断窗口边界是否已经到达的标准。周期默认1s，要修改周期可以调用：

在协同多个输入流，进行窗口化处理时，水位时间戳相当于一个标准时钟。一旦窗口右边界小于等于此时间戳，就会立刻被处理，且小于此时间戳的后续元组均过期。如果多个输入流释放元组的速度差距很大，那么慢的那些流的元组甚至可能全部成为过期元组。

举个例子，假设基于元组字段时间戳的窗口参数如下：

Window length = 20s, sliding interval = 10s, watermark emit frequency = 1s, max lag = 5s

当前墙上时间为09:00:00，在 09:00:00-09:00:01之间接收到以下元组：

e1(6:00:03), e2(6:00:05), e3(6:00:07), e4(6:00:18), e5(6:00:26), e6(6:00:36)

墙上时间到达09:00:01时，水位时间戳06:00:31（最晚元组时间戳-乱序延迟5）被释放，后续任何时间戳小于06:00:31的元组均过期。水位时间戳导致三个窗口被处理：

05:59:50 - 06:00:10  e1 e2 e3
06:00:00 - 06:00:20  e1 e2 e3 e4
06:00:10 - 06:00:30  e4 e5

将第一个元组的时间戳针对滑动间隔向上取整（这确保它落在第一个窗口中），得到06:00:10，作为第一个窗口的右边界。这样有多少个窗口，窗口边界是哪里都可以推算出来。

元组e6没有落在适当的窗口中，因此此时暂不会被处理。

在09:00:01-09:00:02之间接收到以下元组：

e7(8:00:25), e8(8:00:26), e9(8:00:27), e10(8:00:39)

墙上时间戳到达09:00:02时，水位是按戳08:00:34被释放。水位时间戳导致三个窗口被处理：

06:00:20 - 06:00:40  e5 e6
06:00:30 - 06:00:50  e6
08:00:10 - 08:00:30  e7 e8 e9

第一个窗口的左边界，等于上一批次首个窗口的左边界 + 滑动间隔10。第二个窗口继续右滑10s。

向右滑动跳过N多空窗，得到从08:00:10开始的非空窗，包含了e7-e9。e10所属窗口右边界超出水位时间戳，因此暂不处理。

Storm的窗口化功能，可以提供最少一次处理的保证。

在方法 execute(TupleWindow inputWindow)中释放的元组，自动锚定到inputWindow中所有元组。下游Bolt应当Ack从WindowedBolt接收的元组，以完成元组树的处理。如果不Ack元组需要回放，窗口计算也会重新进行。过期的元组会自动被确认。

对于基于时间的窗口，配置参数topology.message.timeout.secs应当远大于windowLength + slidingInterval，否则元组可能来得及处理之前就超时，导致Replay。

对于基于计数的窗口，配置参数topology.message.timeout.secs应当调整，保证windowLength + slidingInterval个元组能够在超时前得到处理。

Storm提供了一个分布式缓存API，利用它可以高效的分发巨大的、在拓扑生命周期内可能变化的文件（或者叫Blob），这些文件的大小可能在KB-GB之间。

对于小的，不需要动态更新的数据集，可以考虑将其嵌入在JAR中。但是过大的文件则不行，会导致拓扑启动非常缓慢。这种情况下使用分布式缓存可以加快拓扑的启动速度。

在拓扑启动时，用户可以指定拓扑所需要的文件集。在拓扑运行期间，用户可以查询分布式缓存中的任何文件，并更换为新版本。更新行为遵从最终一致性模型。在分布式缓存中，文件基于LRU算法管理。Worker负责确认什么文件不再需要，并删除以释放磁盘空间。

分布式缓存的接口是BlobStore，目前它有两种实现：LocalFsBlobStore、HdfsBlobStore

这是基于本地文件系统的缓存实现。

文件的创建由接口ClientBlobStore负责，其实现是NimbusBlobStore。在使用基于本地文件系统的缓存时，NimbusBlobStore调用Nimbus，在其本地文件系统创建文件。

拓扑提交时，JAR和配置文件也是在BlobStore的帮助下上传的，就好像一个普通文件一样。

拓扑启动后，被分配负责运行它的节点的Supervisor会从Nimbus下载JAR、配置文件，并根据topology.blobstore.map确定此拓扑需要使用哪些缓存文件并下载到本地。

在提交拓扑的时候，可以指定该拓扑需要使用哪些文件：

和基于本地文件系统的缓存实现类似，但是容错性已经由HDFS提供了。

LocalFsBlobStore需要在ZooKeeper中存储状态信息，以便支持Nimbus HA。

文件创建、下载、更新由HdfsClientBlobStore负责。

BlobStore基于键、版本来生成一个哈希码，作为Blob的文件名。Blob存放在blobstore.dir指定的目录，默认值对于本地实现来说是storm.local.dir/blobs，对于HDFS实现是一个类似的路径。

一旦Blob被提交，BlobStore即读取配置并为文件创建元数据，元数据用于访问Blob以及对访问者的进行授权。

对于本地实现，Supervisor节点的缓存大小软限制为10240MB，每隔600s会自动根据LRU算法清理超过软限制的那部分blob。

HDFS实现能够减少Nimbus的负载，而且容错性也不受到Nimbus数量的限制。Supervisor进行blob下载时不需要和Nimbus通信，因此减少了对Nimbus的依赖。

Storm的Master是运行在单台机器上的，在监控工具保护下的进程。一般情况下，Nimbus宕机后会由监控工具重启，不会造成太大问题。但是当Nimbus所在机器出现硬件问题时，例如磁盘损坏，Nimbus就无法启动了。

如果Nimbus无法启动，现有的拓扑能够继续运行，但是无法提交新拓扑。现有的拓扑也不能被杀死、激活、禁用。此外，如果某个Supervisor宕掉，Worker重新分配也无法进行。

因此，实现Nimbus的HA是有必要的，以便：

1. 增加Nimbus总体可用时间
2. 允许Nimbus主机在任何时候加入、离开集群
3. 如果Nimbus宕机，不需要进行任何拓扑重提交
4. 活动的拓扑绝不丢失

实现HA，需要在多个Nimbus节点中选择Leader。Nimbus服务器基于如下接口进行Leader选举：

Storm提供了基于ZooKeeper的ILeaderElector实现。

为实现Nimbus故障转移，所有状态、数据必须在所有Nimbus节点之间复制，或者存储在外部的分布式存储系统中。BlobStore就可以用作这样的存储系统。

用户可以通过 topology.min.replication.count声明一个代码复制因子N，在拓扑启动前，其代码、Jar、配置必须复制到至少N个Nimbus节点。使用基于本地文件系统的BlobStore时，当发生故障转移后，如果某个Blob丢失，Nimbus会在需要的时候去下载。也就是说，成为Leader的时候Nimbus不一定在本地存储了所有Blob。

当使用基于本地文件系统的BlobStore时，所有Blob的元数据在ZooKeeper中存储；而使用基于HDFS的BlobStore时，HDFS负责管理。

基于Storm CLI使用分布式缓存的方法，请参考blobstore命令。本节给出相应的Java API的例子：

从0.6.0版本开始，Storm使用Kryo作为串行化库，Kryo是一个高性能、灵活的Java串行化框架。它可以产生较紧凑的串行化格式。

默认情况下，Storm能够串行化基本类型、字符串、字节数组、ArrayList、HashMap、HashSet、Clojure集合类型。如果你希望在元组中使用其它类型，需要提供自定义的串行化器。

在声明输出字段时，Storm不支持指定字段的类型。你仅仅需要把对象放入元组，Storm负责动态的识别其类型并串行化。

这个行为和Hadoop不同，Hadoop对键、值进行强制类型，这需要你提供大量的注解。Hadoop这种强制静态类型往往得不偿失。除了出于简单的目的，下面两条也是Storm选择动态类型的原因：

1. 没有办法很好的静态确定元组字段类型。Storm很灵活，一个Bolt可能订阅了多个流，这些流释放的元组字段各不相同
2. 为了便于和JRuby、Clojure这样的动态语言配合

要自定义串行化器，你需要实现Kryo的某些接口。要注册这些串行化器，你需要配置拓扑的topology.kryo.register，此配置项的值是一个列表：

调用 Config.registerSerialization也可以自定义串行化。

配置项 Config.TOPOLOGY_SKIP_MISSING_KRYO_REGISTRATIONS设置为true时，Storm忽略所有找不到的串行化器类，如果设置为false则抛出异常。

如果Storm遇到某个元组字段，它没有可用的串行化器，则转用Java原生的串行化机制。如果该字段无法基于Java串行化，Storm抛出异常。

注意：Java串行化的成本很高，CPU占用高、输出格式尺寸大。应当尽量避免使用。

配置项 Config.TOPOLOGY_FALL_BACK_ON_JAVA_SERIALIZATION设置为false，则禁用Java串行化。

Storm提供了钩子机制，允许你在任何Storm事件发生时注入自定义代码。扩展BaseTaskHook类，覆盖对应你关心事件的方法即可实现钩子。相关方法包括：emit、spoutAck、spoutFail、boltAck、boltFail、boltExecute。

注册钩子的方法有两种：

1. 在Spout的open、Bolt的prepare方法中，调用TopologyContext的方法注册
2. 通过拓扑配置项topology.auto.task.hooks注册，这样钩子会应用到任何Spout、Bolt。在和外部监控系统集成时，可以使用这种方式

Storm暴露了一个度量（metrics）接口，用于报告整个拓扑的统计信息。在Storm内部，此接口用于元组计数、处理延迟统计、Work堆用量统计，等等。

任何度量都需要实现IMetric接口，该接口仅包含一个方法getValueAndReset，此方法用于执行必要的统计并重置度量为初始值。Storm提供的度量子类型有：

类型	说明
AssignableMetric	支持设置度量为一个明确的值
CombinedMetric	可以被关联更新的度量的通用接口
CountMetric	计数性的度量，其方法 incr()用于增加1个计数， incrBy(n)用于增加n个计数
MultiCountMetric	CountMetric的HashMap
ReducedMetric	MeanReducer：用于求平均值 MultiReducedMetric：ReducedMetric的HashMap

你可以在Bolt中编程式的声明度量、并通过TopologyContext注册：

要注册Worker级别度量，可以配置：

1. Config.WORKER_METRICS：针对集群中所有Worker
2. Config.TOPOLOGY_WORKER_METRICS：针对某个拓扑的所有Worker 

上述两个配置的取值都是度量名到度量类的HashMap。

Worker级别度量具有以下限制：

1. 通过SystemBolt注册，不暴露给用户Task
2. 基于默认构造器创建，不会进行任何属性或配置的注入

要消费度量值实现 IMetricsConsumer接口。通过注册消费者，你可以监听、处理拓扑的度量数据。编程式注册：

通过配置文件：

注册消费者后，Storm会在内部为每个消费者添加一个MetricsConsumerBolt（到拓扑），每个MetricsConsumerBolt都会订阅来自任何Task的度量信息。

MetricsConsumerBolt的并行度由parallelism.hint指定，其组件ID为 __metrics_consumerClassFQName#SEQ，其中SEQ仅仅在多次注册同一个消费者类时出现。

注意：消费者仅仅是一个普通的Bolt，如果它本身性能低下，则很多度量值会因而出现偏差。

Storm提供了一些内置的度量消费者，你可以使用它们来了解拓扑提供了哪些度量：

消费者	说明
LoggingMetricsConsumer	监听度量值，以TSV格式存储到文件
HttpForwardingMetricsConsumer	监听度量值，以HTTP Post方式发送到外部服务器

为了调整Nimbus、Supervisor、运行中拓扑的行为，Storm提供了多种配置信息。 某些配置信息是系统级的，不能为每个拓扑定制。 

所有配置项的默认值记录在defaults.yaml中。要覆盖默认值，在Nimbus、Supervisor的Classpath下放置一个storm.yaml文件。

通过 StormSubmitter提交拓扑时，你可以连同自定义配置项一起提交，这些配置项必须以topology.开头。

从Storm 0.7开始，某些配置项可以在Spout/Bolt级别上定制，包括：

1. topology.debug
2. topology.max.spout.pending
3. topology.max.task.parallelism
4. topology.kryo.register

通过Java API，你可以为组件指定上述配置信息，途径有两种：

1. 组件内部配置：覆盖getComponentConfiguration方法
2. 组件外部配置：调用TopologyBuilder的setSpout/setBolt的返回值的addConfiguration/addConfigurations方法。此会覆盖途径1的配置项

配置信息的优先级，从低到高：defaults.yaml  ⇨ storm.yaml  ⇨ 拓扑配置 ⇨ 组件内部配置 ⇨ 组件外部配置

通过此助手类，可以为拓扑指定配置，此类包含了很多对应了配置项Key的常量值。

要配置一个拓扑在整个集群上，拥有的工作进程的数量，可以使用Config.TOPOLOGY_WORKERS配置项。

要提示为每个组件创建执行线程的数量，可以在setSpout/setBolt时传递parallelism_hint参数，指定初始的执行器数量。

要设置某个组件的任务数量，可以使用Config.TOPOLOGY_TASKS配置项。或者调用 ComponentConfigurationDeclarer.setNumTasks()方法。

调用命令 storm rebalance 可以在运行期间修改拓扑工作进程数量、拓扑组件的执行器数量，示例：

前面我们提到过，Storm支持两种运行模式：本地模式、远程模式。在本地模式中你可以在单JVM进程中调试整个拓扑。

开发环境就是包含了所有Storm组件的环境，你可以在本地模式下开发、测试Storm拓扑。或者将集群提交到远程集群，启动/停止远程集群上的拓扑。

你的开发机器和远程集群的关系是这样的：远程集群被主节点——Nimbus所管理，你的机器和Nimbus交互，提交JAR格式的拓扑。随后拓扑在集群中被调度、执行。你的机器可以通过命令行客户端storm和Nimbus交互，此客户端仅仅支持远程模式。

首先，下载最新版本的Storm，解压后，把bin子目录放到PATH环境变量中。bin目录中包含客户端脚本。

要想管理远程集群，将集群连接信息存放到 ~/.storm/storm.yaml文件中。

主要步骤包括：

1. 创建ZooKeeper集群
2. 在Nimbus、Worker节点上安装依赖
3. 在Nimbus、Worker节点上下载并解压Storm
4. 在storm.yaml中添加必须的配置
5. 使用你选择的监控工具，通过storm命令启动守护程序。

请参考：ZooKeeper学习笔记

安装Java 7+以及Python 2.6.6，不赘述。

Python 3应该也可以工作，但是没有经过详尽的测试。

请参考：搭建开发环境

打开$STORM_HOME/ conf/storm.yaml文件，添加以下配置：

Storm允许你提供一些脚本，让Supervisor定期的执行，以监控节点的健康状态。如果脚本监测到节点不健康，它应当在标准输出打印一行以ERROR开头的文字。Supervisor会周期性的执行脚本，一旦发现ERROR，就会关闭节点上所有Worker进程并退出。

在监控工具下运行守护进程时，你可以考虑调用storm node-health-check命令确认节点健康状态，进一步决定是否继续运行守护进程。

如果需要支持外部库、自定义插件，可以把相关的JAR存放到extlib、extlib-daemon目录下。需要注意， extlib-daemon中的JAR仅仅能被守护进程（不能被Worker也就是拓扑逻辑）使用。

你也可以使用STORM_EXT_CLASSPATH、STORM_EXT_CLASSPATH_DAEMON这两个环境变量。

强烈推荐在监控工具下（supervision tool）启动守护程序，这样意外崩溃后可以立即重启。Storm守护程序被设计为快速失败、无状态的，重启后拓扑不会受到影响：

1. 要启动Nimbus，执行 storm nimbus命令
2. 要启动Supervisor，在每个Worker节点执行 storm supervisor命令
3. 要运行Storm UI服务器，在Nimbus节点执行 storm ui，默认Web端口8080

所有守护程序的日志默认存放在logs目录下。

安装此监控工具参考：Ubuntu下使用monit。服务条目示例：

请参考：官方Docker镜像。 

拉取镜像： docker pull storm:1.1.1

用法示例：

此镜像使用的默认配置是defaults.yaml，改变配置的方式由两种：

1. 使用命令行参数，例如 -c storm.zookeeper.servers='["zookeeper"]'
2. 在容器文件系统路径 /conf/storm.yaml放置配置文件

数据和日志的存储目录位于/data、/logs，这两个目录的所有者为storm。

Storm拓扑的调试主要依赖于日志，特别是在远程集群中运行的情况下。

你可以通过Storm UI或者命令行动态的改变某个包前缀的日志级别，方式类似于设置Log4J的日志级别。

通过Storm UI首页找到需要调试的拓扑，点击链接进入，可以看到类似下面的页面：

此页面提供了和拓扑相关的大量信息：

1. 顶部的搜索栏，可以对Worker的日志进行搜索
2. 按钮区域，提供了激活、禁用、Rebalance、杀死、起停调试功能，还可以设置日志级别
3. 拓扑的名称、状态、Worker数量、Executor数量、Task数量等基本信息
4. 释放、传输（释放并发送给1-N个Bolt）元组的数量、延迟时间，Ack、Fail的数量
5. Spout、Bolt的各类统计信息
6. 拥有的Worker在集群中的分布情况，组件和Worker的对应关系
7. 可以查看图形化的拓扑结构（蓝色表示Spout）
8. 拓扑的配置信息

要通过Storm UI修改日志级别，点击Change Log Level链接，填写表单然后Apply即可。Logger字段可以填写你的代码中使用的日志器前缀（包名前缀）。Timeout指明本次日志级别调整激活多久。

参见命令行客户端。

通过Logviewer不能直接查看Worker的日志文件。Worker的所有日志存放在 ${storm.log.dir}/workers-artifacts/${topologyId}/${port}目录下。worker.log即为工作日志文件，worker.log.err则包含错误级别的日志。

你可以在上述Storm UI界面的顶部，输入关键字（例如Exception）来搜索Worker日志，右上角的放大镜按钮也可以用来进行搜索（默认针对所有拓扑）。

要进行动态JVM剖析，点击拓扑的某个组件，可以看到Profiling and Debugging面板。点选下面的Executors面板的条目，然后点击对应的按钮：

1. JStack：生成Thread Dump
2. Start：启动剖析
3. Heap：生成Heap Dump

点击My Dump Files可以下载上述按钮生成的剖析文件。

配置	说明
worker.profiler.command	指定剖析工具
worker.profiler.enabled	可能被禁用，如果JDK不支持JFR记录、剖析插件也不可用的话

拓扑事件查看器能够用来确定元组如何穿过拓扑的每一个Stage，实时查看元组处理过程时，不会对拓扑运行产生影响。

默认事件记录功能被禁用，你可以设置 topology.eventlogger.executors为大于0的值，以启用之。取值1表示每个拓扑一个日志记录Task，取值nil表示每个Worker一个日志记录Task。

要启动事件记录，点击Topolog Actions面板上的Debug按钮。

要查看被处理的元组，进入对应Spout/Bolt组件的页面，在Component summary面板中，点击events链接。

Storm中的日志记录Bolt利用IEventLogger接口实现事件日志，此接口的默认实现是FileBasedEventLogger。你可以扩展此接口实现更加复杂的逻辑，例如记录到数据库。

Storm提供了一个名为storm的客户端程序，用于和远程集群进行交互。

调用格式：

调用class的main函数，需要的配置文件和依赖的jar包（例如storm的jar），放置于~/.storm目录。main函数通常调用StormSubmitter，并导致topology-jar-path被上传到Nimbus。

如果存在依赖的、没有打包到topology-jar-path的其它jar包，可以使用--jars选项。如果想使用Maven构件的方式指定依赖，可以使用--artifacts选项。

依赖会同时上传作为Worker进程的classpath条目。

调用格式： storm sql sql-file topology-name   --jars  --artifacts --artifactRepositories

编译SQL为Trident拓扑，并提交到Storm集群。

调用格式： storm kill topology-name [-w wait-time-secs]

杀死具有指定名称的拓扑，最多等待wait-time-secs秒。

Storm会首先禁用Spout，然后等待一定时间（默认是拓扑的消息处理超时），让正在处理的消息完成。之后，Storm会关闭Worker进程并清理状态。

调用格式： storm activate topology-name

激活指定拓扑的Spout

调用格式： storm deactivate topology-name

禁用指定拓扑的Spout

调用格式：

某些时候你希望把拓扑的Worker进程均匀分布到更多的机器上。

举例来说，假设你有个10台机器的集群，每个运行4个工作进程。现在，又添加了10个节点，你需要使用这些机器的运算能力，让每个节点运行2个工作进程。一种方法是，杀死并重新提交拓扑，但是rebalance提供了更加简单的方法。

rebalance首先禁用指定的拓扑Spout（等待到消息超时或者-w），然后，在集群中均匀的创建Worker进程，最后激活拓扑。

rebalance也可以用来改变拓扑的并发度： -n用来改变Worker进程数量；-e用来指定组件的Executor数量（线程数量，和Task数量不是一回事）

调用格式： storm repl

开启Clojure REPL环境。

调用格式： storm classpath，打印客户端的classpath信息。

调用格式： storm localconfvalue conf-name

打印配置项conf-name的本地值，本地配置是~/.storm/storm.yaml合并了defaults.yaml的结果。

调用格式： storm remoteconfvalue conf-name

必须在集群中的节点上执行，打印集群的配置项。集群配置是$STORM-PATH/conf/storm.yaml合并了defaults.yaml的结果。

调用格式： storm nimbus。启动Nimbus守护进程，此命令应该在 daemontools、 monit之类的supervision工具中调用。

调用格式： storm supervisor。启动Supervisor守护进程，此命令应该在 daemontools、 monit之类的supervision工具中调用。

调用格式： storm ui。启动Web UI服务器守护进程，此命令应该在 daemontools、 monit之类的supervision工具中调用。

调用格式： storm drpc。启动DRPC服务器守护进程，此命令应该在 daemontools、 monit之类的supervision工具中调用。

调用格式： storm blobstore cmd。管理Storm的分布式缓存。子命令列表：

子命令	说明
list [KEY...]	列出当前存储的blob
cat [-f FILE] KEY	读取一个blob，写入到文件或者stdout
create [-f FILE] [-a ACL ...]     [--replication-factor NUMBER] KEY	创建一个blob，其内容来自文件或者stdin ACL是一个逗号分隔的列表，条目格式[uo]:[username]:[r-][w-][a-]
update [-f FILE] KEY	更新一个blob的内容
delete KEY	删除一个blob
set-acl [-s ACL] KEY	设置一个blob的访问控制列表
replication --read KEY	读取blob的复制因子

onitor
调用格式： storm dev-zookeeper

启动一个全新的zookeeper服务器，使用dev.zookeeper.path作为其本地目录，使用storm.zookeeper.port作为其端口。仅仅用于开发/测试。

调用格式： storm get-errors topology-name

获得指定拓扑的最新的错误信息。返回结果是component-name: component-error的键值对JSON

调用格式： storm heartbeats [cmd]

调用格式： storm kill_workers

杀死运行在此Supervisor节点上的Worker进程，必须在Supervisor节点上调用。

调用格式： storm list。列出集群中运行的拓扑，以及这些拓扑的状态。

调用格式： storm logviewer。启动日志查看器进程，此进程提供Web UI，此命令应该在 daemontools、 monit之类的supervision工具中调用。

调用格式：

交互式的对指定拓扑的吞吐量进行监控。 

调用格式： storm node-health-check。在本地Supervisor上运行健康状态检测。

调用格式： storm pacemaker。启动Pacemaker进程，此进程提供Web UI，此命令应该在 daemontools、 monit之类的supervision工具中调用。

调用格式：

动态的控制拓扑的日志级别。

调用格式： storm upload_credentials topology-name [credkey credvalue]*

更新credentials集到运行中的拓扑。

Trident（音标/'traɪd(ə)nt/）是Storm提供的另外一种访问接口，它提供一些高层抽象，包括：

1. 精确的一次处理语义
2. 事务性数据持久化
3. 一些通用的流分析功能

使用Trident你可以无缝的将低延迟分布式查询和高吞吐量（每秒百万级消息）、有状态流处理混合。如果你对Pig、Cascading之类的高层批处理工具熟悉，理解起Trident会很容易。Trident有类似的连接（Join）、聚合（Aggregation）、分组（Grouping）、函数（Function）、过滤器（Filter）概念，此外Trident还提供在任何数据库、存储机制基础上进行有状态、增量处理的原语。

考虑两个流处理需求：

1. 从输入流中统计每个姓名出现的次数
2. 能够传入一个姓名列表，获得这些姓名出现的总次数

非线性随机名字生成器：