首页 > 代码库 > Strom优化指南

Strom优化指南

        摘要:本文主要讲了笔者使用Strom中的一些优化建议

1、使用rebalance命令动态调整并发度

Storm计算以topology为单位,topology提交到Storm集群中运行后,通过storm rebalance 命令可对topology进行动态调整。比如增加Topology的worker数,修改Bolt,Spout的并行执行数量 parallelism等,从而实现topology的动态调整,达到弹性计算的目的。(当然调整时要配合监控模块)

基本上主要有两种用法:
1) storm rebalance topology-name -n new-work-num,
调整指定topology的worknum。
2)storm rebalance topology-name -e component=parallelism
调整指定topology中指定component的并行数量.

注:Jstorm不提供这个功能

2、使用tick消息做定时器

使用Storm组件的定时器需要为bolt重写下面的方法:
public Map<String, Object> getComponentConfiguration() {
Map<String, Object> conf = new HashMap<String, Object>();
conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS, 60);//每60s持久化一次数据
return conf;
}
其中:Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS 定时消息发送的频率,单位为秒。
我们判断是否为tick消息。可以使用TupleHelpers类中的isTickTuple方法,具体代码:
public static boolean isTickTuple(Tuple tuple) {
return tuple.getSourceComponent().equals(Constants.SYSTEM_COMPONENT_ID) && tuple.getSourceStreamId().equals(
Constants.SYSTEM_TICK_STREAM_ID);
}

3、使用组件的并行度代替线程池

    道理很简单,通过给bolt/spout 设置parallelism,而不是在bolt/spout设置线程池。这样可以避免一个组件将一台机器资源耗尽。

4、不要在Spout中处理耗时操作

   Spout都是单线程的,如果太多的耗时操作就在这里,那么整个程序的处理吞吐量就会下降。而且如果nextTuple方法非常耗时,那么消息会成功执行完毕后,Acker给Spout发送消息,Spout如果无法及时消费,可能造成Ack消息超时丢弃。然后Spout就认为这个消息执行失败

5、单个spout/bolt中做一件事

   每个spout/bolt只做一件事,比如消息的第一次清洗放在spout,接下来的计算放在bolt,再接下来计算结果入库再放到一个bolt,将整个任务串连多个spout/bolt起来。而不是单独的放在一个组件中。这样才能最大使用集群的资源。也好方便对一个单独的组件进行调优

6、注意fieldsGrouping的数据均衡性

   Strom中有6种fields方法,fieldsGrouping是按字段进行分组.通过合理的设置。确实各个Bolt/spout接收的消息都比较均衡。避免单个节点机器处理大量数据,这样耗时又耗机器。这里就涉及到一个field字段的选择。

7、优先使用LocalOrShuffleGrouping

     数据首先优先选择本节点上的bolt处理,减少不必要的数据传输。其它Storm Grouping如下,

Shuffle Grouping :随机分组,尽量均匀分布到下游Bolt中
将流分组定义为混排。这种混排分组意味着来自Spout的输入将混排,或随机分发给此Bolt中的任务。shuffle grouping对各个task的tuple分配的比较均匀。
Fields Grouping :按字段分组,按数据中field值进行分组;相同field值的Tuple被发送到相同的Task
这种grouping机制保证相同field值的tuple会去同一个task,这对于WordCount来说非常关键,如果同一个单词不去同一个task,那么统计出来的单词次数就不对了。“if the stream is grouped by the “user-id” field, tuples with the same “user-id” will always go to the sae task”
All grouping :广播
广播发送, 对于每一个tuple将会复制到每一个bolt中处理。
Global grouping :全局分组,Tuple被分配到一个Bolt中的一个Task,实现事务性的Topology。
Stream中的所有的tuple都会发送给同一个bolt任务处理,所有的tuple将会发送给拥有最小task_id的bolt任务处理。
None grouping :不分组
不关注并行处理负载均衡策略时使用该方式,目前等同于shuffle grouping,另外storm将会把bolt任务和他的上游提供数据的任务安排在同一个线程下。
Direct grouping :直接分组 指定分组
由tuple的发射单元直接决定tuple将发射给那个bolt,一般情况下是由接收tuple的bolt决定接收哪个bolt发射的Tuple。这是一种比较特别的分组方法,用这种分组意味着消息的发送者指定由消息接收者的哪个task处理这个消息。 只有被声明为Direct Stream的消息流可以声明这种分组方法。而且这种消息tuple必须使用emitDirect方法来发射。消息处理者可以通过TopologyContext来获取处理它的消息的taskid (OutputCollector.emit方法也会返回taskid)。

8、不是所有场景都要使用ACK机制

     如果说处理的消息每天都有上亿,消息丢失个几百几千其实对计算结果的影响是非常小的。先说一下ACK机制:

 为了保证数据能正确的被处理, 对于spout产生的每一个tuple, storm都会进行跟踪。这里面涉及到ack/fail的处理,如果一个tuple处理成功是指这个Tuple以及这个Tuple产生的所有Tuple都被成功处理, 会调用spout的ack方法;如果失败是指这个Tuple或这个Tuple产生的所有Tuple中的某一个tuple处理失败, 则会调用spout的fail方法;
在处理tuple的每一个bolt都会通过OutputCollector来告知storm, 当前bolt处理是否成功。
另外需要注意的,当spout触发fail动作时,不会自动重发失败的tuple,需要我们在spout中重新获取发送失败数据,手动重新再发送一次。
Ack原理

  Storm中有个特殊的task名叫acker,他们负责跟踪spout发出的每一个Tuple的Tuple树(因为一个tuple通过spout发出了,经过每一个bolt处理后,会生成一个新的tuple发送出去)。当acker(框架自启动的task)发现一个Tuple树已经处理完成了,它会发送一个消息给产生这个Tuple的那个task。
Acker的跟踪算法是Storm的主要突破之一,对任意大的一个Tuple树,它只需要恒定的20字节就可以进行跟踪。
Acker跟踪算法的原理:acker对于每个spout-tuple保存一个ack-val的校验值,它的初始值是0,然后每发射一个Tuple或Ack一个Tuple时,这个Tuple的id就要跟这个校验值异或一下,并且把得到的值更新为ack-val的新值。那么假设每个发射出去的Tuple都被ack了,那么最后ack-val的值就一定是0。Acker就根据ack-val是否为0来判断是否完全处理,如果为0则认为已完全处理。

使用IBasicBolt接口实现自动确认

为了简化编码,Storm为Bolt提供了一个IBasicBolt接口,它会在调用execute()方法之后正确调用ack()方法,BaseBasicBolt类是该接口的一个实现,用来实现自动确认

9、尽量抽取公用的处理部分到一个组件

     比如说存储到数据库的动作。可以尽量都使用同个bolt来写,管理好线程池

10、合理设置work数目

      work数目并不是越多越好,还要看你的数据和你的处理逻辑。这个一般情况下可以反复调整参数来确定最优,可以通过查看页面上的各个bolt/spout处理消息的耗时,资源使用情况来确定

11、GC参数优化

     对每个work的jvm参数进行调整,推荐生产使用parNew+CMS垃圾回收方式。

    上面就是笔者自己的一些总结,希望对你有帮助。

Strom优化指南