storm第一篇--概念，例子，参数优化

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(); 
        }
    }
}

运行结果效果如下：

转自：http://www.cnblogs.com/linjiqin/archive/2013/05/28/3104016.html

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的数目