Storm性能优化

2018-11-21 本文已影响68人尼小摩

性能优化1：kryo序列化

定制序列化

自定义的bolt之间emit数据是实体类的时候，注册kryo
Storm 使用 Kryo 来处理序列化。如果要实现自定义的序列化生成器，需要注册一个新的 Kryo 的序列化生成器。

可以通过拓扑的 topology.kryo.register 属性来添加自定义序列化生成器。两种姿势：

只有一个待注册的类的名称。在这种情况下，Storm 会使用 Kryo 的 FieldsSerializer 来序列化该类。conf.registerSerialization(UserEntity.class);
一个包含待注册的类的名称和对应的序列化实现类名称，该序列化实现类实现了 com.esotericsoftware.kryo.Serializer接口。

topology.kryo.register:
  - com.ly.CustomType1
  - com.ly.CustomType2: com.ly.serializer.CustomType2Serializer
  - com.ly.CustomType3

Storm序列化文章：
https://www.cnblogs.com/intsmaze/p/7044042.html
http://storm.apache.org/releases/1.0.6/Serialization.html

性能优化2：KafkaBolt批量提交

props.put("request.required.acks", "0");
props.put("producer.type", " async");
props.put("message.send.max.retries", "3");�props.put("batch.num.messages", "200");
props.put("send.buffer.bytes", "1024*100");

性能优化3：使用组件的并行度代替线程池

Storm 自身是一个分布式、多线程的框架，对每个Spout 和Bolt，我们都可以设置其并发度；它也支持通过rebalance 命令来动态调整并发度，把负载分摊到多个Worker 上。如果自己在组件内部采用线程池做一些计算密集型的任务，比如JSON 解析，有可能使得某些组件的资源消耗特别高，其他组件又很低，导致Worker 之间资源消耗不均衡，这种情况在组件并行度比较低的时候更明显。
如某个Bolt 设置了1 个并行度，但在Bolt 中又启动了线程池，这样导致的一种后果就是，集群中分配了这个Bolt 的Worker 进程可能会把机器的资源都给消耗光了，影响到其他Topology 在这台机器上的任务的运行。如果真有计算密集型的任务，我们可以把组件的并发度设大，Worker 的数量也相应提高，让计算分配到多个节点上。

性能优化4：注意fieldsGrouping 的数据均衡性

fieldsGrouping 是根据一个或者多个Field 对数据进行分组，不同的目标Task 收到不同的数据，而同一个Task 收到的数据会相同。
假设某个Bolt 根据用户ID 对数据进行fieldsGrouping，如果某一些用户的数据特别多，而另外一些用户的数据又比较少，那么就可能使得下一级处理Bolt 收到的数据不均衡，整个处理的性能就会受制于某些数据量大的节点。可以加入更多的分组条件或者更换分组策略，使得数据具有均衡性。

性能优化5：优先使用localOrShuffleGrouping

localOrShuffleGrouping 是指如果目标Bolt 中的一个或者多个Task 和当前产生数据的Task 在同一个Worker 进程里面，那么就走内部的线程间通信，将Tuple 直接发给在当前Worker 进程的目的Task。否则，同shuffleGrouping。localOrShuffleGrouping 的数据传输性能优于shuffleGrouping，因为在Worker 内部传输，只需要通过Disruptor 队列就可以完成，没有网络开销和序列化开销。因此在数据处理的复杂度不高，而网络开销和序列化开销占主要地位的情况下，可以优先使用localOrShuffleGrouping 来代替shuffleGrouping。

性能优化6：设置合理的Worker 数

Worker 数越多，性能越好？先看一张Worker 数量和吞吐量对比的曲线

从图可以看出，在12 个Worker 的情况下，吞吐量最大，整体性能最优。这是由于一方面，每新增加一个Worker 进程，都会将一些原本线程间的内存通信变为进程间的网络通信，这些进程间的网络通信还需要进行序列化与反序列化操作，这些降低了吞吐率。
另一方面，每新增加一个Worker 进程，都会额外地增加多个线程（Netty 发送和接收线程、心跳线程、SystemBolt 线程以及其他系统组件对应的线程等），这些线程切换消耗了不少CPU，sys 系统CPU 消耗占比增加，在CPU 总使用率受限的情况下，降低了业务线程的使用效率。