Strom优化指南

Posted 2020-09-06 Evankaka

        摘要：本文主要讲了笔者使用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垃圾回收方式。

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