storm_Helloword案例

【准备】
java环境
python环境
zookeeper环境
storm环境

上代码：
一般的storm程序大概分为三个组件（spout，bolt，topology），当然可以有多个spout，或者多个bolt。

spout：
    数据源，它是storm要解析数据的入口
bolt：
    数据业务处理组件（可以理解为数据在bolt中流动并接受业务处理）
topology：
    拓扑。它的编写主要实现数据流向顺序，组装spout和bolt。一个拓扑打包了一个实时处理程序的逻辑

【Spout】（spout的实现方式主要有两种：继承 BaseRichSpout 实现 IRichSpout ）

/**
 * 
 * @author mis
 * 
 *         创建spout类PWSpout，数据源 
 *          实现方式一:继承BaseRichSpout 
 *          实现方式二:实现IRichSpout 
 */

public class PWSpout extends BaseRichSpout {

    private static final long serialVersionUID = 1l;
    // 数据源是往出写数据的
    private SpoutOutputCollector collector;
    // 制造数据源
    private static final Map<Integer, String> map = new HashMap<Integer, String>();
    //模拟数据
    static {
        map.put(0, "java");
        map.put(1, "php");
        map.put(2, "groovy");
        map.put(3, "python");
        map.put(4, "ruby");
    }

    //初始化方法
    public void open(Map conf, TopologyContext context,
            SpoutOutputCollector collector) {
        // 对spout进行初始化
        this.collector = collector;
    }

    // 轮询方法，发射数据
    public void nextTuple() {
        // 随机发送一个单词
        final Random r = new Random();
        int num = r.nextInt(5);
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {}
        // 发射数据
        this.collector.emit(new Values(map.get(num)));// 解决发射声明，发射内容

    }

    // 指明
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        // 进行声明
        declarer.declare(new Fields("print"));// 发往数据的id，解决下一个节点去哪取发射的数据
    }
}

【Bolt】（bolt的实现方式主要有两种：继承 BaseBasicBolt 实现 IRichBolt）

【bolt1组件】
/**
 * 
 * @author mis
 * 创建的一个Bolt：PrintBolt接收数据并计算或者发射给下一个节点
 * 
 *  实现方式一：继承BaseBasicBolt
 *  实现方式二：实现IRichBolt
 */
public class PrintBolt extends BaseBasicBolt {

private static final Log log = LogFactory.getLog(PrintBolt.class);
    
    private static final long serialVersionUID = 1l;

    //接收数据进行业务操作操作
    public void execute(Tuple input, BasicOutputCollector collector) {
        //获取上一个组件所声明的Field，这里的id与上一个spout发送数据的id要对应才能获取spout发送的数据
        String print = input.getStringByField("print");
        //发送给下一个bolt
        collector.emit(new Values(print));  
    }

    //声明数据id
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("write"));  //可以理解为这个bolt将数据发向id为write的bolt组件
    }
}

【bolt2组件】
/**
 * 
 * @author mis 创建的一个Bolt：WriteBolt接收数据并计算或者发射给下一个节点
 * 
 *         实现方式一：继承BaseBasicBolt 
 *         实现方式二：IRichBolt
 */
public class WriteBolt extends BaseBasicBolt {

    private static final long serialVersionUID = 1l;

    private static final Log log = LogFactory.getLog(WriteBolt.class);

    private FileWriter writer;

    @Override
    public void execute(Tuple input, BasicOutputCollector collector) {
        // 获取上一个组件所声明的Filed
        String text = input.getStringByField("write");
        String sysname = System.getProperty("os.name");
        
        try {
            if (writer == null) {
                if(sysname.equals("Windows 10") || sysname.equals("Windows 8.1") || sysname.equals("Windows 7")){
                    writer = new FileWriter("D:\\storm_helloword\\" + this);
                }else if (System.getProperty("os.name").equals("Linux")) {
                    writer = new FileWriter("/usr/local/temp/stormData/" + this);
                }
            }
            log.info("【write】:写入文件");
            writer.write(text);
            writer.write("\n");
            writer.flush();
        } catch (Exception e) {}
    }

    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        //这里数据处理结束的话就不用发送至下一个bolt组件
    }
}

【Topology】实时处理程序的逻辑

/**
 * 
 * @author mis
 * 组件逻辑，可以理解为各个组件执行顺序的规定类
 */
public class PWTopology1 {

    public static void main(String[] args) throws Exception {
        Config cfg = new Config();
        //cfg.setNumWorkers(2);
        cfg.setDebug(true);
        
        //组件执行顺序逻辑
        TopologyBuilder builder = new TopologyBuilder();
        //先创造spout数据源PWSpout
        builder.setSpout("spout", new PWSpout());
        //在创建bolt组件PrintBolt，并接受分组id为“spout”的数据
        builder.setBolt("print-bolt", new PrintBolt()).shuffleGrouping("spout");
        //在创建bolt组件WriteBolt，并接受分组id为“print-bolt”的数据
        builder.setBolt("write-bolt", new WriteBolt()).shuffleGrouping("print-bolt");
        
        
        //1 本地模式
        LocalCluster cluster = new LocalCluster();
        //提交分析
        cluster.submitTopology("top1", cfg, builder.createTopology());
        //10000毫秒后停止实时计算
        Thread.sleep(10000);
        cluster.killTopology("top1");
        cluster.shutdown();
        
        //2集群模式(任务id以参数的形式传入)
        //StormSubmitter.submitTopology(args[0], cfg,builder.createTopology());
    }
}

方案一

2017-04-25_155610.png

方案二

2017-04-25_162634.png

方案三

2017-04-25_162652.png

以上三种方案代表不同的组件处理逻辑

【本地运行】

2017-04-25_160041.png

【集群打包运行】

打包上传jar
启动相关程序
./zkServer.sh start
storm nimbus &
storm supervisor &
storm ui &

2017-04-25_161713.png

storm jar /root/Desktop/stormHelloword.jar storm01.topology.PWTopology1 top2
                        jar包地址                      指定执行主类       传入任务id

【storm的jar终端提交命令】

启动：storm jar  【jar路径】 【拓扑包名.拓扑类名】 【拓扑名称】
关闭：storm kill 【拓扑名称】

2017-04-25_162430.png

更详细的任务信息

2017-04-25_162912.png