1 概念
目前最新的0.8.0版本里面
worker -> 进程。一个worker只能执行同一个spout/bolt的task，一个worker里面可以有多个executor。
executor -> 线程。 一个executor执行可以执行多个task。
task -> storm进行任务分配的基本单位。
2 例子
storm实战入门一
本节探讨一下storm具体怎么使用，明白怎么在windows下开发storm程序。
功能描述：实时随机输出一字符串。
在开发前记得导入storm需要的jar包。
1、SimpleSpout类继承BaseRichSpout类，用来产生数据并且向topology里面发出消息：tuple。
package com.ljq.helloword; import java.util.Map; import java.util.Random; import backtype.storm.spout.SpoutOutputCollector; import backtype.storm.task.TopologyContext; import backtype.storm.topology.OutputFieldsDeclarer; import backtype.storm.topology.base.BaseRichSpout; import backtype.storm.tuple.Fields; import backtype.storm.tuple.Values; /** * Spout起到和外界沟通的作用，他可以从一个数据库中按照某种规则取数据，也可以从分布式队列中取任务 * * @author Administrator * */ @SuppressWarnings("serial") public class SimpleSpout extends BaseRichSpout{ //用来发射数据的工具类 private SpoutOutputCollector collector; private static String[] info = new String[]{ "comaple ,12424,44w46,654,12424,44w46,654,", "lisi ,435435,6537,12424,44w46,654,", "lipeng ,45735,6757,12424,44w46,654,", "hujintao ,45735,6757,12424,44w46,654,", "jiangmin ,23545,6457,2455,7576,qr44453", "beijing ,435435,6537,12424,44w46,654,", "xiaoming ,46654,8579,w3675,85877,077998,", "xiaozhang ,9789,788,97978,656,345235,09889,", "ceo ,46654,8579,w3675,85877,077998,", "cto ,46654,8579,w3675,85877,077998,", "zhansan ,46654,8579,w3675,85877,077998,"}; Random random=new Random(); /** * 初始化collector */ public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) { this.collector = collector; } /** * 在SpoutTracker类中被调用，每调用一次就可以向storm集群中发射一条数据（一个tuple元组），该方法会被不停的调用 */ @Override public void nextTuple() { try { String msg = info[random.nextInt(11)]; // 调用发射方法 collector.emit(new Values(msg)); // 模拟等待100ms Thread.sleep(100); } catch (InterruptedException e) { e.printStackTrace(); } } /** * 定义字段id，该id在简单模式下没有用处，但在按照字段分组的模式下有很大的用处。 * 该declarer变量有很大作用，我们还可以调用declarer.declareStream();来定义stramId，该id可以用来定义更加复杂的流拓扑结构 */ @Override public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declare(new Fields("source")); //collector.emit(new Values(msg));参数要对应 } }
2、SimpleBolt类继承BaseBasicBolt类，处理一个输入tuple。
package com.ljq.helloword; import backtype.storm.topology.BasicOutputCollector; import backtype.storm.topology.OutputFieldsDeclarer; import backtype.storm.topology.base.BaseBasicBolt; import backtype.storm.tuple.Fields; import backtype.storm.tuple.Tuple; import backtype.storm.tuple.Values; /** * 接收喷发节点(Spout)发送的数据进行简单的处理后，发射出去。 * * @author Administrator * */ @SuppressWarnings("serial") public class SimpleBolt extends BaseBasicBolt { public void execute(Tuple input, BasicOutputCollector collector) { try { String msg = input.getString(0); if (msg != null){ //System.out.println("msg="+msg); collector.emit(new Values(msg + "msg is processed!")); } } catch (Exception e) { e.printStackTrace(); } } public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declare(new Fields("info")); } }
3、SimpleTopology类包含一个main函数，是Storm程序执行的入口点，包括一个数据喷发节点spout和一个数据处理节点bolt。
package com.ljq.helloword; import backtype.storm.Config; import backtype.storm.LocalCluster; import backtype.storm.StormSubmitter; import backtype.storm.topology.TopologyBuilder; /** * 定义了一个简单的topology，包括一个数据喷发节点spout和一个数据处理节点bolt。 * * @author Administrator * */ public class SimpleTopology { public static void main(String[] args) { try { // 实例化TopologyBuilder类。 TopologyBuilder topologyBuilder = new TopologyBuilder(); // 设置喷发节点并分配并发数，该并发数将会控制该对象在集群中的线程数。 topologyBuilder.setSpout("SimpleSpout", new SimpleSpout(), 1); // 设置数据处理节点并分配并发数。指定该节点接收喷发节点的策略为随机方式。 topologyBuilder.setBolt("SimpleBolt", new SimpleBolt(), 3).shuffleGrouping("SimpleSpout"); Config config = new Config(); config.setDebug(true); if (args != null && args.length > 0) { config.setNumWorkers(1); StormSubmitter.submitTopology(args[0], config, topologyBuilder.createTopology()); } else { // 这里是本地模式下运行的启动代码。 config.setMaxTaskParallelism(1); LocalCluster cluster = new LocalCluster(); cluster.submitTopology("simple", config, topologyBuilder.createTopology()); } } catch (Exception e) { e.printStackTrace(); } } }
运行结果效果如下：
3 参数优化
1 并行度
摘要：workers为storm提供的工作进程，程序的并行度可以设置（包括spout和bolt的并行度，如果有acker的话还包括acker的并行度）并行度即为task 数目。
一般而言 worker和task之间的比例,即1个worker包含10~15个左右,当然根据配置和应用需要测试优化。
一个storm topology运行起来之后, 会在supervisor 机器上启动一些进程来运行spout和bolt实例.
如果一个topology里面一共有一个spout, 一个bolt。 其中spout的parallelism是2, bolt的parallelism是4, 那么我们可以把这个topology的总工作量看成是6， 那么一共有6个task，那么/tasks/{topology-id}下面一共会有6个以task-id命名的文件，其中两个文件的内容是spout的id,
其它四个文件的内容是bolt的id。
task->node+port, 它其实就是从task-id到supervisor-id+port的映射， 也就是把这个task分配给某台机器的某个端口来做。
topology里面的组件(spout/bolt)都根据parallelism被分成多个task, 而这些task被分配给supervisor的多个worker来执行。
task都会跟一个componment-id关联, componment是spout和bolt的一个统称.
对于每一个component在部署的时候都会指定使用的数量, 在storm-user中有一个讨论说明了这个问题:
里面的大意是说, 通过设置parallelism来指定执行spout/bolt的线程数量. 而在配置中还有另外一个地方(backtype.storm.Config.setNumWorkers(int))来指定一个storm集群中执行topolgy的进程数量, 所有的线程将在这些指定的worker进程中运行.
比如说一个topology中要启动300个线程来运行spout/bolt, 而指定的worker进程数量是60个, 那么storm将会给每个worker分配5个线程来跑spout/bolt, 如果要对一个topology进行调优, 可以调整worker数量和spout/bolt的parallelism数量(调整参数之后要记得重新部署topology. 后续会为该操作提供一个swapping的功能来减小重新部署的时间).
对于worker和task之间的比例, nathan也给出了参考,
即1个worker包含10~15个左右, 当然这个参考, 实际情况还是要根据配置和测试情况
转自 http://xumingming.sinaapp.com/category/storm/
2 内存优化
Storm中真正干活的是各个worker，而worker由supervisor负责启动。在topology启动过程中我们会看到如下的启动日志：
这就是启动一个worker进程，也就是一个JVM进程。
默认情况下，Storm启动worker进程时，JVM的最大内存是768M。
但我在使用过程中，由于会在Bolt中加载大量数据，768M内存无法满足需求，会导致内存溢出程序崩溃。
经过研究发现，可以通过在Strom的配置文件storm.yaml中设置worker的启动参数:
worker.childopts: "-Xmx2048m"该参数会在启动时传递给JVM，然后就可以在worker中使用2048m内存了。
目前好像Storm还没有配置文件的详细说明，比如可以配置哪些参数，怎么配置？
大家可以先参考Storm源代码中的Config.java.
3 其他
max_spout_pending=100000 //spout可以缓存的tuple数目
topology_message_timeout_secs=60 // message被完全处理完的时间，超过这个时间，spout 将产生failed
num_workers=4 //worker的数目