Storm深度分析及其正式版本思考

Storm发展到现在已经有了5个年头，从刚开始惊艳四方，到现在逐渐被新兴框架(Flink、Spark Streaming)挑战。Storm本身也在不断的发展，Twitter对其不断的探索，且深一步的开发了Heron框架。社区也在憋了5年后发布了第一个正式版本。

Storm内部机制及探索

内部机制

Storm写了一层调度系统，Nimbus作为调度的Master(类似ResourceManager)，Supervisor作为工作机器上的监控进程(类似NodeMonitor)，Worker作为真正的工作进程，使用ZooKeeper作为中间的通信手段(不同于YARN直接使用RPC的方式)，系统更健壮，就算Nimbus挂了也不会影响其它进程的运行，但是也带来了规模化的问题，ZK毕竟撑不住太大的流量。

主要的通信流程是：Supervisor把自己资源情况心跳汇报给Nimbus(ZK)、Worker把健康情况心跳给supervisor(文件)、Worker把健康及stat信息心跳给Nimbus(ZK)、Nimbus把分配信息通知给Supervisor和Worker(ZK)、Supervisor下载用户的作业资源(Nimbus Thrift)。

Nimbus主要管理作业的资源分配，Jar包分发、对外提供的管理接口。

主要线程有：

1.分配线程：拿到所有作业，根据其需要分配的worker数(有可能新提交上来，也有可能Rebalance，也有可能executor心跳超时了)，将supervisor心跳上来的资源分配给Worker。

2.清理线程：作业资源清理/Supervisor超时清理。

3.Thrift服务线程池。

Supervisor主要作两件事：同步分配(包括下载资源)、启停Worker。

(图来自Twiter发的Storm论文)

为了不让下载资源等耗时操作影响轮训间隔，Supervisor使用一个专门的Timer发送周期性事件，并且进行心跳。

为了让同步分配后马上进行启动Worker(也有可能是分配变了杀掉worker)，在进行一次同步分配后，会直接发送一个启停worker的事件。

Worker的设计主要牵涉到了数据流的传输，在实时计算中，数据的无延迟传输非常重要，设计上必须是Record-By-Record形式。

(图来自Twiter发的Storm论文)

接收数据：Netty层收到数据会放到LinkedBlockingQueue里，接收线程从里取出数据，分发到各个Executor的接收队列里，用户逻辑线程反序列化数据，并执行用户逻辑(在storm1.0中已经 取消了接收线程，并且由Netty线程直接进行反序列化)。

发送数据：用户线程把要发送的数据放到自己的发送队列里，对应的发送线程把数据拿出来，序列化放到全局发送队列里，由一个专门的全局发送线程发给Netty层。

可以看出，Storm为了各个线程间无影响，使用了大量的队列，这虽然保证了实时性，但是造成了吞吐量不高，线程利用率低的问题(Storm1.0中设计了Batch DisruptorQueue，大大增加了吞吐量)。

规模化问题

ZooKeeper在Storm承载了太多了任务，心跳(其中还包括Worker的stat指标)、分配的同步。可以说，Storm本身就不是为了大规模集群(机器数大于200台)而设计的。但是在生产环境中，需求总是越来越多的，这就导致需要搭建的Storm集群会越来越多，几十个上百个集群，这给运维带来了灾难性的后果。

所以还是需要把单个Storm集群做大，Twitter在运维的时候进行了各种尝试。

刚开始为了ZooKeeper的吞吐量，将ZooKeeper的日志和Snapshot的磁盘分离，这样尽可能让它们之间无干扰，以防有瓶颈。

第二点是控制KafkaSpout的读写ZK，它需要把自己当前读取队列的偏移持久化，可以适当调整频率(也可以换成HBase等)。

第三点是要把集群做大，还是不应该把最重的心跳放ZK上，可以把它放在一个专门的心跳服务上去(Storm1.0版本提供了Pacemaker服务来作心跳)。

其实如果解决了On Yarn问题后，直接走RPC给Nimbus也是可以的。

Tuning难问题

Storm在生产中，作业很难真正Run起来，其中一部分原因就是很难进行调优，用户需要设置worker数、并发数、以及更难的Spout Max Pending，设小了会导致吞吐上不去，设大了会导致worker OOM，Twitter设计了一个自动调优的Auto Tuning算法。

大致思路是根据一段时间内Ack回来的数据量进行逐步调整MaxPending值，会有一定波动，直到一个比较稳定的值，这件事情说起来简单，但是实际生产上却不一定能比较好的Run起来，因为实际环境会有各种各样的其他因素导致的不稳定，算法是不是能健壮的Run起来，这很考验算法开发者的工程能力。

Storm1.0：重大进步

性能优化

提高了6-10倍，在大吞吐情况甚至大大减小了数据延时。主要在三个方面进行了优化：

Batch优化

在与Spark Streaming和Flink的对比中，Storm各种被诟病吞吐低，就此，社区主要就打包进行了各项研究尝试。主要有Tuple打包和队列打包两条分支，分别是：

1) STORM-855: Add tuple batching

2) STORM-1151: Batching in DisruptorQueue

Tuple打包改动较大，而且不易控制，更容易造成数据的震荡，而队列打包影响小，并且是直接针对Storm的瓶颈所在进行改造，storm原有瓶颈大多都是因为DisruptorQueue的严重竞争导致。经过各种测试，队列打包在各种场景，吞吐量和延时都要优于Tuple打包。

线程模型优化

Batch优化过后，发送者足够快时，这时瓶颈主要是在接收者这边，Storm1.0把Worker接收线程去掉，然后由Netty线程直接把数据反序列化后发送到executor队列中(netty线程不会被阻塞，因为这个队列是优化过后的DisruptorBatchQueue)，这样带来的效果：减少线程与队列搬移，把反序列化从executor线程中提取出来。

Clojure代码优化

主要分析执行过程中方法调用栈的每一步耗时，优化其中耗时的使用反射的clojure代码。

Spout反压

受Heron刺激了，主要是根据接收队列和全局发送队列的水位线来通过ZK通知Spout，将其置为deactive。

有震荡，当然效果不如逐级反压。

(图来自Storm设计文档)

Pacemaker作心跳服务

解决ZK规模化问题。

分布式缓存

Nimbus提供接口，存储在Nimbus机器文件系统中或HDFS中，如果在作业中制定了缓存资源，Supervisor会在启动worker前把资源下载好，worker可以直接使用此资源。

supervisor会清理失效缓存，使用LRU算法和一个超时时间。

HA Nimbus

早就该做了!这里是热备，通过ZK抢主。

Window Api

还是通过以前的Bolt基础上提供的。

State Checkpoint

有点类似Flink中Master把checkpoint消息“插入”到正常的数据流中的做法，Storm这里的做法没有Barrier，所以也没有完全分布式一致的checkpoint，所以只能做到At-Least-once。

(图来自Storm文档)

只是at-least-once，还是通过acker机制来做，数据量不大时可行。

Trident Window，更有实际意义，现语义还比较简单。

Resource Aware Scheduler

作业分配按照CPU和内存的资源方面来进行更有针对的分配，可以下一步就是CGroup隔离。

动态日志等级

通过ZK来做了一些动态参数。

Tuple Sampling and Debugging

可以动态抽样数据进行调试，这点在调试意义上是非常方面的，动态的通过ZK通知Worker，使Worker在发数据时会抽样额外发送一份到一个专门的Debug Bolt节点，这个节点会写到文件中，然后通过logview进程来查询提供给UI。

分布式日志搜索

对作业所有日志进行搜索，主要也是为了调试方便。

动态worker的Profiling

获取jstack/heap dump等。

仍然存在的缺点

没有成熟支持YARN

规模化问题解得不彻底

Storm1.0吞吐已经有很大提高，但是CPU利用率还是较低。

毕竟是真正的Record-By-Record。

Trident

虽然解决追踪消息过多/Exactly-Once问题，但是还有缺点：

1.难配参数batchSize/batchInterval，虽然有反压，但是Spout反压不如逐级反压。

2.batchInterval使得数据处理的延时固定了，类似Spark Streaming的划小批，不能达到实时计算的效果。

3.虽然有窗口了，但是支持不如flink(比较好的支持DataFlow中的窗口语义)。

4.commit过于频繁，导致性能瓶颈在外部存储上(不可以按时间checkpoint)。

5.强依赖HBase，且当状态内存装不下时，退步到全HBase的操作(不支持RocksDb+HDFS)，比如Distinct。

6.直接使用Kryo序列化低效(不如Writable)+大吞吐量时队列数据堆积导致GC严重，时间久进入老年代。