首页 > 安全资讯 >

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

16-08-16

Storm发展到现在已经有了5个年头,从刚开始惊艳四方,到现在逐渐被新兴框架(Flink、Spark Streaming)挑战。Storm本身也在不断的发展,Twitter对其不断的探索,且深一步的开发了Heron框架。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严重,时间久进入老年代。

相关文章
最新文章
热点推荐