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. 能够传入一个姓名列表，获得这些姓名出现的总次数

非线性随机名字生成器：

Storm代码： 

Trident以较小的批次的方式来处理输入流，根据输入throughput的不同，一个批次包含的元组数量可以在数千到数百万之间。 

Trident提供了全特性的API，用于处理元组批次。这些API和Pig、Cascading等Hadoop高层抽象很类似。Trident提供了跨越批次进行聚合、持久化存储（可以在内存、Redis等“状态源”中）聚合结果的功能。此外，Trident还具有查询实时状态源的功能，这些状态可以被Trident更新，状态源也可以是独立于Trident维护的。

回到上面的人名统计代码中：

1. Spout释放包含单个字段names的元组，names字段实际上是空格分隔的人名列表
2. 我们使用Split函数对每个names字段进行分割，每个输入元组可能产生多个输出元组，输出元组包含单个字段name。Split函数的实现：
   
   Java

   1 2 3 4 5 6 7 8 9

   public class Split extends BaseFunction {    // 处理一个输入元组，可以释放任意个输出元组    public void execute(TridentTuple tuple, TridentCollector collector) {        String sentence = tuple.getString(0);        for(String word: sentence.split(" ")) {            collector.emit(new Values(word));                        }    } }

3. 我们使用groupBy对人名进行分组，相同的名字分为一组
4. 我们使用聚合器Count对分组进行聚合，并调用persistentAggregate对聚合结果进行持久化。此方法知道如何存储、更新位于状态源中的聚合结果。在此例中我们使用内存作为状态源，实际项目中你可以使用Memcached、Cassandra、Redis等持久化存储，例如：
   
   Java

   1	.persistentAggregate(MemcachedState.transactional(serverLocations), new Count(), new Fields("nameCount"));

   注意：persistentAggregate所存储的值，是输入流发出的所有批次的总的聚合结果 

5. persistentAggregate把流转换为TridentState对象。在本例中，TridentState对象包含了所有人名的计数结果。使用此TridentState对象可以实现需求的分布式查询部分

Trident的一个很酷的地方是，它提供完全的容错、精确的一次处理语义，这可以很好的简化实时处理逻辑。通过巧妙的存储状态，Trident能够在出错而需要重试时，不会对相同的源数据进行重复的入库（例如聚合到Memcached）。

人名统计需求的第一部分已经实现，拓扑运行后，TridentState不断更新，记录了当前各人名出现的总次数。

需求的第二部分是实现对统计结果的低延迟、实时查询。对于一个空格分隔的人名列表，我们要返回这些名字出现的次数的总和。

可以看到，查询客户端和普通的RPC客户端没有区别， 只是它的请求可以在Storm集群中并行的处理。类似于这种小的查询请求，其延迟可以低达10ms。

我们需要对上面的拓扑进行扩展，让它作为DRPC服务器：

查询服务器业务逻辑说明：

1. 每个DRPC请求被作为一个独立的小批处理任务来看待，整个请求参数被作为单个元组处理。此元组仅仅包含一个args字段，对应客户端传递来的请求参数字符串
2. 首先，Split函数对请求参数进行分割
3. 操作符stateQuery能够对TridentState对象进行查询。TridentState对象由拓扑前一部分的persistentAggregate调用暴露 —— 持久化后的聚合结果允许后续的查询操作。stateQuery接受三个参数：
   1. 状态源，在本例中即持久化的人名统计结果
   2. 用于查询的函数，在本例中是MapGet，输入字段name作为函数入参
   3. 查询结果输出字段
4. 上一步的返回值是一个流，我们用FilterNull这个过滤器进行过滤，把空值去掉
5. 最后，调用聚合器Sum进行聚合，结果仍然是一个流。此流会被Trident自动发送给等待结果的客户端

本例中的拓扑需要提交到Storm集群才能运行：

下面是一个纯粹的DRPC拓扑，用于计算按需计算URL的Reach值。Reach值针对单个URI，统计暴露于此URI的用户的数量。具体算法是，针对给定的URI，获取所有推特了此URI的用户、以及这些用户的Follower，并将所有这些用户去重、计数。

计算Reach对于单台机器来说，计算密集度太高了，它可能上千个数据库查询、上千万的元组处理。利用Trident可以很轻松的实现并行处理。

该DRPC拓扑拓扑会读取两个状态源：一个映射URI到推特了此URI的用户的列表；另外一个映射用户到其Follower的列表：

Trident能够智能的决定如何执行拓扑，让性能最大化。在上面的例子中，我们应当注意一些有趣的细节：

1. 对于需要对状态源进行读、写的操作（persistentAggregate、stateQuery），会自动的批量式的和源交互。如果当前批次需要对状态源进行20个更新操作，Trident会自动将这20个操作进行批量提交，不管是20个写还是读。也就是说Trident仅仅和状态源进行一次交互
2. Trident的聚合器是高度优化的，它不会把分组中所有的元组传递给单台机器，然后进行聚合。它会尽可能的多执行局部聚合，避免跨越网络发送不必要的流量。例如聚合器Count会在每个节点上进行部分计数，把部分计数通过网络发送，最终获得总合计数。这种行为特点和MapReduce的combiners类似

Trident数据模型由TridentTuple表示，和普通Storm元组一样，它也是命名值的列表。

在Trident拓扑中，操作节点通常将若干输入字段作为参数，并释放一系列函数字段（function Fields），这些新字段被添加到元组中。

考虑一个流stream，释放三元素x、y、z构成的元组，下面的过滤器仅仅把y字段作为输入：

上述过滤器将仅保留y小于10的那些元组。 再看看下面的函数：

该函数以x、y字段为输入字段，同时定义了两个输出字段 —— 这些输出字段是添加到输入元组上的，这导致输入元组现在有5个字段：x、y、z、added、multiplied。

使用聚合器的情况下，输出字段则是代替了输入字段，此外输入输出元组也不是一一对应关系：

使用分组+聚合器的情况下，输出字段同时包含分组所依据的字段：

类似于原始Storm API，Trident的数据来源也是Spout。

你可以在Trident中使用原始的Spout，这种Spout是非事务性的：

要定制ZooKeeper相关参数，修改transactional.zookeeper.*配置项。 

Trident拓扑的默认行为是，每次处理一个批次。当前批次成功或者失败后，才可能处理下一个批次。

通过管线化批次的处理流，可以大大提升拓扑的吞吐能力、降低单个批次的处理延迟。要定制同时能够处理的批次的最大数量，修改topology.max.spout.pending参数。

即使同时处理多个批次，Trident的严格有序语义保持不变 —— 状态更新总是串行化、有序的执行。以单词统计为例，当Batch 1进行全局聚合的时候，Batch 2...10不能进行全局聚合，但是它们可以进行分区的部分聚合。

Trident提供以下专有类型的Spout：

Spout类型	说明
ITridentSpout	最通用的接口，可以用于实现事务性、非透明事务性语义
IBatchSpout	每次释放一个元组批次的非事务性Spout
IPartitionedTridentSpout	从分区数据源（例如Kafka集群）读取数据的事务性Spout
IOpaquePartitionedTridentSpout:	从分区数据源（例如Kafka集群）读取数据的非透明事务性Spout

实时计算的一个关键问题是，如何管理状态——包括外部数据源的状态、拓扑的内部状态——以确保当失败、重试发生时，对状态的更新是幂等的。 在生产环境中，失败是常态，当节点宕机或者出现其它问题，只能重试处理元组批次。所谓幂等，可以简单的理解为，每个元组批次导致的状态更新，仅仅发生一次。

Trident确保幂等性的机制是：

1. 每个元组批次被赋予唯一性的ID，所谓事务ID。如果发生重试，批次的事务ID保持不变
2. 不同批次之间的状态更新，是严格有序的。也就是说，Batch3对状态源的更新动作不会发生，直到Batch2的状态更新动作完成

这种处理机制和Storm中的事务性拓扑类似。但是，你不需要像事务性拓扑那样，自己记录txid、比较txid、存储多个数据项，这些工作已经由Trident的状态抽象封装起来了。

你可以选择任何期望的策略来存储状态。例如存放在外部数据库中、存放在内存中并以HDFS作为后备。

状态信息也非必须永久保存，你可以考虑使用内存状态存储，并仅仅保留最近N个小时的数据。

可能和容错相关的Spout有三类：非事务性的、事务性的、不透明事务性的，相应的，对应了三类状态对象：

Spout	说明
事务性Spout	事务性Spout有以下特性： 每个批次的txid总是保持不变。Replay时，批次包含的元组完全不变 不同批次的元组没有交集 任何元组都属于一个批次 这是最理想、简单的情况。例如storm-contrib提供的支持Kafka的事务性Spout实现：TransactionalTridentKafkaSpout 事务性Spout对应的状态实现，利用了txid和元组集的绝对对应关系。用一个例子说明，现在有一个Trident拓扑，用于统计单词出现数量，并持久化到一个键（单词）值（计数）数据库中。为了支持事务性，仅仅存储键值对是不够的，你必须知道哪些批次已经入库。由于Trident批次的严格有序性，额外引入一个值字段，记录上一次处理完成的txid就可以了。当Replay时，如果发现库中txid和当前批次的txid一样，跳过入库步骤即可 更具体化一些，假设txid 3包含元组： ["man"] ["man"] ["dog"]，当前数据库状态如下： Perl 1 2 3 4 # 一定要保证，状态和txid原子性的入库 man => [count=3, txid=1]        # After [count=5, txid=3] dog => [count=4, txid=3]        # After [count=4, txid=3] apple => [count=10, txid=2] 可以看到，单词man计数3，上一次成功的txid不等于当前txid3，因此应当把本批次的man计数2入库。再强调一下，由于批次的严格有序性，只要库中的txid不等于当前txid，那么它必然是前面批次的txid而绝不会是后面的。再看看dog记录的txid和当前txid一致，因此本批次dog计数1不应该入库 为什么上一次提交出现dog入库成功，man入库失败的情况呢？这牵涉到底层数据源的工作机制，我们假设它不支持原子的写两个键值 那么需要担心txid 没有处理完成么？不需要，还是严格有序性问题，一旦txid3执行到更新状态源这一步了，说明前面所有txid都成功完成了
不透明事务性Spout	为什么不仅仅使用事务性Spout？原因是，保证完全的容错可能需要更苛刻的外部条件，而这种容错级别可能并非必要 以TransactionalTridentKafkaSpout为例，它的一个批次中的元组，可能来自Kafka主题的所有分区。一旦某个批次被释放，为了满足事务性语义，未来它处理失败需要Replay时，所有分区都必须可读。如果某个Kafka节点宕机（假设没有启用Replication）了呢？那么分区就读不到，相应的就无法满足事务性，整个Trident拓扑就卡住了 不透明（Opaque）事务性拓扑用于解决上述问题。它允许丢失某些数据源节点，但是仍然保证一次性处理语义 不透明事务性拓扑的特性是： 不保证每个批次的元组完全一致 保证每个元组仅仅在一个批次中被成功处理。一个元组可能在Batch 1中失败，后来在Batch 3中成功 OpaqueTridentKafkaSpout是不透明事务性Spout的例子，它允许Kafka主题的某个分区节点临时宕机。不管什么时候，它总是从上一批次最后一个Kafka记录（对应元组）的偏移量处读取下一个批次。这确保每个记录仅仅被一个批次成功处理 使用不透明事务时，记录txid字段的技巧不再有效，因为批次Replay时它包含的元组可能不同。但是，引入更多的状态字段，仍然可以达成容错目标。仍然看单词计数的例子，假设当前状态源的数据如下： Perl 1 2 3 4 5 6 man => (     // 对于每个单词，存储三个字段，这些字段仍然需要原子的更新     value => 4,        // 本次事务更新后的值     prevValue => 1,    // 本次事务更新前的值     txid => 2          // 本次事务ID ) 现在来了包含元组 ["man"] ["man"]的txid2，也就是和数据库记录的txid相同。这意味着什么？txid2的上一次尝试失败了。由于同一事务的两个批次的元组集可能不同，因此上次尝试记录的value是无效的。应当以prevValue为基准加上本批次包含的元组数量2，成功处理后数据变为： Perl 1 man => ( value => 3, prevValue => 1,txid => 2 ) 然后，又来了包含元组 ["man"] ["man"]的txid3，也就是和数据库记录的txid不同。这意味着上一次批次处理成功了，因此数据库中记录的value是有效的。应当以value为基准加上本批次包含的元组数量2，成功处理后数据变为： Perl 1 man => ( value => 5, prevValue => 3,txid => 3 ) 这种机制的关键仍然是严格有序性，一旦txid3准备入库，就可以确信txid2是成功的，因此其value就有效，否则，必须回滚为prevValue再行计算
非事务性Spout	不对批次中包含的内容作任何保证

三种Spout必须搭配对应类型的状态实现使用：

其中非透明事务性状态具有最强的容错能力，但是要为每个状态数据存储额外的两个字段。事务性状态仅需要存储一个额外的字段，但是它只能和事务性Spout搭配使用。

上节讨论的复杂的存储细节，并不需要开发者自行实现。Trident已经把它们全部封装到状态API中了。

调用partitionPersist、persistentAggregate等方法，即可创建一个状态对象，例如： 

状态（源）是一个接口，它很简单：

Trident不对你的状态的工作方式做任何假设，它不规定调用什么方法可以更新状态，调用什么方法可以读取状态。

假设有一个存储用户位置的数据库，你想通过Trident访问它，可以这样实现State：

为了支持状态的查询，Trident提供了QueryFunction接口。 为了支持状态的更新，Trident提供了StateUpdater接口。状态和状态的支持的操作被解耦，如此设计是因为Trident难以对特定状态源的工作方式作出合理假设。

沿用上面的例子，QueryFunction的实现示例如下：

使用上述QueryFunction的代码示例：

StateUpdater的实现示例如下：

使用上述StateUpdater的代码示例：

除了partitionPersist，Trident提供的另外一类可以更新状态源的API是persistentAggregate。此API是建立在partitionPersist上的一层抽象，它知道如何使用Trident聚合器，将聚合后的数据更新到状态源。

你可以针对GroupedStream调用persistentAggregate，参数必须提供MapState接口的实现。此接口的签名如下：

如果要对非分组流进行（全局性）聚合，State对象必须实现下面的接口：

MemoryMapState、MemcachedState 实现了 MapState、Snapshottable接口。

当实现自己的MapState时，你不需要从零开始。Trident提供了OpaqueMap、TransactionalMap、NonTransactionalMap类，并把容错相关逻辑封装其中。你需要做的是，为这三个类提供IBackingMap实现：

Trident提供了基于LRU缓存算法的IBackingMap实现：CachedMap，继承它以实现具有缓存特性的状态源。

Trident提供了SnapshottableMap，可以把MapState转变为Snapshottable，其实就是全局使用单个key进行分组。

Trident拓扑会被尽可能编译为高效的Storm拓扑，仅仅在发生数据重分区（例如洗牌、分组）时，元组才通过网络传输。

一个可能的Trident拓扑和Storm拓扑的对应关系如下图：

可以看到，Storm以分区节点为界，从Trident拓扑划分出多个Bolt。这些Bolt内部可能包含了多个Trident节点，它们可能被分配到一个Worker中执行，避免了不必要的网络数据拷贝。

Trident的核心数据模型是流，流由一系列的元组批次构成。流可能跨越集群的多个节点而分区，应用到流的操作针对每个分区并行执行。

Trident的操作包含五种类型：

1. 在分区本地运行，不牵涉到网络流量的
2. 重新分区操作，仅仅对流进行重新分区，不改变流的内容
3. 聚合操作，产生一定的网络流量
4. 针对分组（Grouped）流的操作
5. 合并、连接操作

操作	说明
Function	函数根据一系列输入字段，经过计算产生0-N个输出字段。这些输出字段被添加到原始的输入元组中，构成新的字段 如果函数没有释放新字段，则对应的元组被过滤掉 要创建函数，考虑继承BaseFunction，要在Trident中使用函数，调用流的each方法
Filter	过滤器以元组为输入，决定它是否被保留 要创建过滤器，考虑继承BaseFilter，要在Trident中使用过滤器，可以调用流的filter方法
map	针对每个元组进行1:1的转换，输出为转换后的元组 要创建Map函数，考虑继承MapFunction，要在Trident中使用它，调用map方法
flatMap	针对每个元组进行1:N的转换，输出为转换后的元组 要创建flatMap函数，考虑继承FlatMapFunction，要在Trident中使用它，调用flatMap方法
peek	用于针对每个元组进行额外的操作，例如查看每个元组如果经过处理管线。可以用于调试目的 peek不会影响流的后续处理
project	仅仅保留选中的输入元组字段，示例： Java 1 2 // 仅仅保留b、d两个字段 stream.project(new Fields("b", "d"))
min / minBy	对一个元组批次的每一个分区，返回该分区中指定字段的值最小的那个元组。你可以自定义比较器函数
max / maxBy	对一个元组批次的每一个分区，返回该分区中指定字段的值最大的那个元组。你可以自定义比较器函数
Windowing	后文详述
partitionAggregate	针对批次的每个分区进行内部的聚合操作，其输出替换掉输入元组，示例： Java 1 2 stream.partitionAggregate(new Fields("price"), new Sum(), new Fields("sum")); // 输出元组包含一个字段
stateQuery	查询状态源
partitionPersist	更新状态源

这类操作运行一个函数，来决定分组如何在不同Task之间分配。重新分区后，分区数量可能改变，你可以调用 parallelismHint()改变并行度亦即分区数量。

重新分区操作可能导致网络流量，支持的重新分区函数有：

重新分区操作	说明
shuffle	随机轮回方式将元组均匀的分配给所有分区
broadcast	将每个元组广播（重复）到所有分区
partitionBy	根据一系列字段进行语义分区，先根据字段值计算哈希，然后针对分区数量取模，取模结果相同的元组被发给同一分区
global	所有元组被发送给单个分区，所有批次也都发送给该分区
batchGlobal	所有元组被发送给单个分区，不同批次可以发送给不同分区
partition	传入一个重分区函数（实现CustomStreamGrouping接口）进行分区操作

Trident提供两个全局性（针对所有分区）的聚合操作：

聚合操作	说明
aggregate	针对每个批次（的所有分区）进行聚合操作
persistentAggregate	针对流的所有批次进行聚合操作，并且将结果存放在状态源

分组流操作	说明
groupBy	该操作产生一个GroupedStream —— 根据某些字段，把字段值相同的元组分到一个区里面 也就是说，该操作依据字段进行重新分区操作 针对GroupedStream执行聚合操作时，聚合在每个分区（组）内部进行，而不是针对整个批次 针对GroupedStream调用persistentAggregate时，结果存储到MapState对象中，其键为分组所依据的字段 类似于普通流，GroupedStream也支持聚合的链式调用

这些定义在TridentTopology上的API用于把不同的流合并到一起：

操作	说明
merge	将多个流合并为一个，示例： topology.merge(stream1, stream2, stream3); 合并产生的新流，其字段根据第一个输入流的字段决定
join	类似于SQL的连接操作。当连接发生在来自不同Spout的流之间时，这些Spout将被同步 —— 每个批次将包含来自每个Spout的元组 如果要实现窗口化连接（Windowed Join） —— 例如，来自流A的元组仅仅后过去一小时产生的流B的元组进行连接，可以考虑利用partitionPersist、stateQuery，最近一小时的数据以轮换（Rotated）方式存储在状态源中，并且以连接字段为键。这样，stateQuery就很容易查询到最近一小时的数据了 示例： Java 1 2 3 4 5 6 7 8 // userStream包含字段 id, personId, loginName // personStream包含字段 id, personName topology.join(     userStream, new Fields("personId"), personStream, new Fields("id"),     // 连接后的流的字段，第一个字段对应用于连接的那个字段     // 后续一次是前一个、后一个流中，不属于连接字段的所有字段，按照顺序逐一对应     new Fields("personId","userId","loginName","personName") );

注意：针对全局性聚合使用：

1. ReducerAggregator或者Aggregator时，流首先被重分区到单个分区，然后聚合器在此新分区上执行
2. CombinerAggregator时，首先进行局部聚合，然后重分区到单个分区，最后完成聚合操作

局部聚合可以大大减少网络流量，因此应该尽可能使用CombinerAggregator。

partitionAggregate、aggregate、persistentAggregate等操作，在调用时需要传入聚合器。聚合器相关的接口有三种：

聚合器接口	说明
CombinerAggregator	接口签名： Java 1 2 3 4 5 6 7 8 public interface CombinerAggregator<T> extends Serializable {     // 针对每个输入元组，进行初始化，产生一个值     T init(TridentTuple tuple);     // 每个元组产生的值，一次被送入下面的函数进行reduce，直到仅剩一个值     T combine(T val1, T val2);     // 如果没有任何输入元组，调用此函数     T zero(); } 此聚合器返回单个字段构成的单一元组作为输出 进行aggregate（而非partitionAggregate）时，此聚合器接口的优势体现在，Trident能够自动优化，尽可能的进行局部聚合，避免发送网络流量
ReducerAggregator	接口签名： Java 1 2 3 4 5 6 public interface ReducerAggregator<T> extends Serializable {     // 初始值     T init();     // 针对每个输入元组进行reduce     T reduce(T curr, TridentTuple tuple); } 可以和persistentAggregate一起使用
Aggregator	这个是最一般化的接口： Java 1 2 3 4 5 6 7 8 public interface Aggregator<T> extends Operation {     // 在处理每个批次之前，调用返回值是一个代表了聚合状态的对象，会被传递给后面两个方法     T init(Object batchId, TridentCollector collector);     // 针对批次分区中每个输入元组调用。此方法会更新state，还可以释放元组     void aggregate(T state, TridentTuple tuple, TridentCollector collector);     // 当批次分区中所有输入元组都被处理之后调用此方法     void complete(T state, TridentCollector collector); }

某些情况下，你需要连续执行多个聚合器，参考如下方式：

Trident支持窗口化（Windowing），即对元组批次中，处于同一窗口中的那些元组进行处理，并释放聚合后的结果给下一个操作。

两种风格的窗口被支持：

1. 滚动窗口（Tumbling window）—— 元组依据单个窗口分组，分窗要么基于处理时间要么基于元组计数。每个元组只能属于一个窗口（不重叠）：
   
   Java

   1 2 3 4 5 6 7 8 9 10 11 12 13

   /** * 返回一个新流，新流元组由当前流窗口内元组基于aggregator聚合而成 * 窗口为滚动窗口，每隔windowCount个元组，新建一个窗口 */ public Stream tumblingWindow(int windowCount, WindowsStoreFactory windowStoreFactory,     Fields inputFields, Aggregator aggregator, Fields functionFields);   /** * 返回一个新流，新流元组由当前流窗口内元组基于aggregator聚合而成 * 窗口为滚动窗口，每隔windowDuration这么长时间，新建一个窗口 */ public Stream tumblingWindow(BaseWindowedBolt.Duration windowDuration, WindowsStoreFactory windowStoreFactory,     Fields inputFields, Aggregator aggregator, Fields functionFields);

2. 滑动窗口（Sliding window）—— 分窗依据和上面类似，但是每个一定的间隔（处理时间或元组计数），窗口就向前滑动。元组可以属于多个窗口（重叠）：
   
   Java

   1 2 3 4 5 6 7 8 9 10 11

   /** * 滑动窗口，每个窗口包含windowCount个元组，每处理slideCount个元组则窗口向前滑动 */ public Stream slidingWindow(int windowCount, int slideCount, WindowsStoreFactory windowStoreFactory,     Fields inputFields, Aggregator aggregator, Fields functionFields);   /** * 滑动窗口，每个窗口时长为windowDuration，每经过slidingInterval则窗口向前滑动 */ public Stream slidingWindow(BaseWindowedBolt.Duration windowDuration, BaseWindowedBolt.Duration slidingInterval,     WindowsStoreFactory windowStoreFactory, Fields inputFields, Aggregator aggregator, Fields functionFields);

除了上面两套API外，Trident还支持一个通用的窗口API：

Trident的资源感知调度器（Resource Aware Scheduler，RAS）API可以用于限制Trident拓扑的资源消耗。此API和基本的Storm RAS API类似，区别只是该API针对Trident流而非Bolts、Spouts调用。示例：

 

1. 元组处理代码要尽可能轻量，execute方法中的代码要能够尽快执行完毕。当扩容部署时，性能影响会成倍增加
2. 当内存需求较小（1GB以下）时，考虑使用Guava实现Bolt本地缓存
3. 当需要跨越多个Bolt共享数据时，考虑使用HBase、Phoenix、Redis或者Memcached
4. 尽可能将配置信息外部化到属性文件中，可以避免运维期间不必要的代码重编译

要防止拓扑重部署后，消息被重复处理。使用KafkaSpout时，要确保：

1. SpoutConfig的id、zkroot在重部署后要保持不变。Storm使用这这两个属性决定偏移量在ZooKeeper中的存储路径
2. 注意在生产环境中设置forceFromStart为false。如果设置为true，KafkaSpout会忽略ZooKeeper中存储的偏移量信息，从头开始消费。开发环境下经常设置为true，以重复测试一批消息
3. 如果需要部署同一拓扑的多个版本，则应当分配不同的ClientID，Kafka将不同ClientID作为单独Consumer看待，可以独立的跟踪偏移量

1. 工作进程的个数，设置为节点（机器）数量的整数倍。并行度设置为工作进程数量的整数倍。kafka分区的数量设置为Spout并行度的整数倍
2. 考虑为每个(拓扑,机器)组合启动单个Worker进程
3. 启动较少较大的聚合组件，这类组件在Worker节点上运行一个实例
4. 每个Worker使用一个Acker，0.9开始这是默认值
5. 监控JVM垃圾回收器活动，如果一切正常，应该很少出现Major GC
6. 设置Trident的批量处理延迟为你的应用的端对端延迟的1/2
7. 将Spout最大未决元组数量的初值设置的很小，对于Trident可以设置为1或者拓扑执行器线程数量。然后逐步增加设置值，直到流运行状况稳定。最终到达的值可能接近于：\[2 \times \frac{throughputInSecs}{sec} \times endToEndLatency \]

Worker进程的总数由Supervisor来设置。每个Supervisor都管理若干JVM Slot。你在拓扑上设置的参数，实际上是声明Slot的数量。

对于每个(拓扑,机器)的组合，没有太大必要使用超过1个的Worker。对于一个运行在3个8核节点上的拓扑，并行能力提示是24，每个Bolt在每个机器上配备8个执行器线程。3工作进程 x 8执行器线程的方案，比起24工作进程的方案，有三个优势：

1. 数据被重新分区（shuffle/group-bys）时，分发给的执行器如果在同一进程内，不需要进程间传递数据。本机进程间传递数据缓存需要Work发送 - 本地套接字 - Work接收，尽管不需要使用网卡，仍然比进程内直接调用慢
2. 对于执行聚合功能的组件来说3个使用大后备缓存（backing cache）的组件比起24个使用小缓存的组件，具有更小的effect of skew和更高的LRU效能
3. 较少的Worker进程，意味着较少的控制逻辑流量

1. 检查是否对日志目录具有写权限
2. 检查JVM的堆大小
3. 检查是否所需的库都安装到Worker进程
4. 检查ZooKeeper连接配置
5. 是否正确的为每个Worker节点设置了唯一性机器名，并且写到Storm配置文件中
6. 检查防火墙，允许所有Worker节点、Master节点、ZooKeeper之间的双向通信

典型症状：本地模式下运行正常，但是部属到集群中后工作节点启动即崩溃

可能原因：子网的配置可能有问题，导致节点无法根据hostname相互发现。有时ZeroMQ会因为无法解析主机名而崩溃

解决方案：

1. 在hosts文件中配置主机名和IP地址的映射关系
2. 使用内部DNS服务器

典型症状：

1. 所有的元组均处理失败
2. 处理流程不工作

解决方案：

1. 注意Storm不支持IPv6，可以考虑设置Supervisor的 Child JVM系统属性-Djava.net.preferIPv4Stack=true
2. 子网配置可能有问题，参考上一个问题

典型症状：

1. 一开始处理正常，但是一段时间后，突然停止处理，Spout的元组开始集体fail

解决方案：

1. 可能和ZeroMQ版本有关，可以从2.1.10降级为2.1.7

典型症状：

1. 某些Supervisor不出现在Storm UI中
2. Storm UI刷新后，Supervisor列表发生变化

解决方案：

1. 确保每个Supervisor的本地目录是独立的，不要通过NFS共享本地目录
2. 尝试删除Supervisor的本地目录并重启守护程序。由于Supervisor会为自己创建一个UUID并存放在本地，如果此ID复制到其它节点，Storm集群会出现问题

一定要保证构建JAR时使用的Storm版本和集群使用的Storm版本一致

输出元组中的对象必须实现不变模式。一旦你释放元组到OutputCollector，你就不应该在修改它。

在Storm生命周期中，拓扑会被串行化并存储到ZooKeeper中。这意味着Spout/Bolt必须支持串行化，包括它们的属性。如果某些属性无法支持串行化，考虑在Spout/Bolt的prepare方法中初始化它。这些方法在拓扑组件被分发给Worker进程后才会执行。

报错信息：Caused by: java.lang.RuntimeException: java.io.IOException: Found multiple defaults.yaml resources. You're probably bundling the Storm jars with your topology jar. [jar:file:/apache-storm-1.1.1/storm-local/supervisor/stormdist/name-counter-trident-topology-2-1508312849/stormjar.jar!/defaults.yaml, jar:file:/apache-storm-1.1.1/lib/storm-core-1.1.1.jar!/defaults.yaml]

报错原因：打包时把Storm的JAR包也打到拓扑的JAR包中会导致这个情况的发生。

配置拓扑时，需要指明DRPC服务器的位置：