Strom优化指南

摘要:
文摘:本文主要介绍Strom中的一些优化建议。1.使用重新平衡命令动态调整并发性。风暴计算以拓扑为单位。将拓扑提交到Storm集群以供运行后,可以通过stormred平衡命令动态调整拓扑。Spout的并行运行次数是并行的,从而实现拓扑的动态调整,达到弹性计算的目的。Spout假设它不能及时消耗,这可能会导致ACK消息在超时后被丢弃。只有声明为DirectStream的消息流才能声明此类分组方法。

        摘要:本文主要讲了笔者使用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、使用组件的并行度取代线程池

    道理很easy,通过给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优化指南》仅用于学习参考。如对内容有疑问,请及时联系本站处理。

上篇下拉菜单(css)电商系统设计下篇

宿迁高防,2C2G15M,22元/月;香港BGP,2C5G5M,25元/月 雨云优惠码:MjYwNzM=

相关文章

【转载】Apache Storm 官方文档 —— 基础概念

【转载自https://yq.aliyun.com/articles/87510?t=t1】 Storm 系统中包含以下几个基本概念: 拓扑(Topologies) 流(Streams) 数据源(Spouts) 数据流处理组件(Bolts) 数据流分组(Stream groupings) 可靠性(Reliability) 任务(Tasks) 工作进程(W...

k8s zookeeper安装(集群版与非集群版)

集群版zookeeper安装 第一步:添加helm镜像源 helm repo add incubator http://storage.googleapis.com/kubernetes-charts-incubator 第二步:下载Zookeeper helm fetch incubator/zookeeper 第三步:修改 ... persiste...

流式处理框架storm浅析(上篇)

本文来自网易云社区 作者:汪建伟 前言 前一段时间参与哨兵流式监控功能设计,调研了两个可以做流式计算的框架:storm和spark streaming,我负责storm的调研工作。断断续续花了一周的时间看了官网上的doc和网络上的一些资料。我把所学到的总结成一个文档,发出来给对storm感兴趣的同事做入门引导。 storm背景 随着互联网的更进一步发...

阿里的STORM——JSTORM

看介绍文档貌似挺好:https://github.com/alibaba/jstorm 阿里拥有自己的实时计算引擎 类似于hadoop 中的MR 开源storm响应太慢 开源社区的速度完全跟不上Ali的需求 降低未来运维成本 提供更多技术支持,加快内部业务响应速度 现有Storm无法满足一些需求 现有...

Storm常见模式——分布式RPC

Storm常见模式——分布式RPC 本文翻译自:https://github.com/nathanmarz/storm/wiki/Distributed-RPC,作为学习Storm DRPC的资料,转载必须以超链接形式标明文章原始出处及本文翻译链接。 分布式RPC(distributed RPC,DRPC)用于对Storm上大量的函数调用进行并行计算过程。...

实时计算,流数据处理系统简介与简单分析

转自:http://www.csdn.net/article/2014-06-12/2820196-Storm 摘要:实时计算一般都是针对海量数据进行的,一般要求为秒级。实时计算主要分为两块:数据的实时入库、数据的实时计算。今天这篇文章详细介绍了实时计算,流数据处理系统简介与简单分析。 编者按:互联网领域的实时计算一般都是针对海量数据进行的,除了像非实时...