storm 入门学习实例

storm 入门学习实例,第1张

storm 入门学习实例

前段时间,项目中使用到了流式处理方面的技术,学习了一下storm,编写了一个小实例。

1.引入jar包



    4.0.0
    
        org.springframework.boot
        spring-boot-starter-parent
        2.3.12.RELEASE
         
    
    com.example
    control
    0.0.1-SNAPSHOT
    control
    Demo project for Spring Boot
    
        1.8
    
    

        
            org.springframework.boot
            spring-boot-starter
            
                
                    log4j-to-slf4j
                    org.apache.logging.log4j
                
                
                    jul-to-slf4j
                    org.slf4j
                
                
                    logback-classic
                    ch.qos.logback
                
            
        


        
            org.springframework.boot
            spring-boot-starter-web
        


        
            org.apache.storm
            storm-core
            2.2.1
        

        
            com.codahale.metrics
            metrics-core
            3.0.2
            
                
                    slf4j-api
                    org.slf4j
                
            
        

    

    
        
            
                org.springframework.boot
                spring-boot-maven-plugin
                
                    
                        
                            org.projectlombok
                            lombok
                        
                    
                
            

            
                org.apache.maven.plugins
                maven-shade-plugin
                
                    
                        package
                        
                            shade
                        
                    
                
                
                    false
                    
                        
                            commons-logging:commons-logging
                            javax.servlet:servlet-api
                            javax.mail:javax.mail-api
                        
                    
                
            

        
    


2.编写程序:

(1)编写输入流类
package test.storm;


import java.util.Map;
import java.util.Random;


import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.baseRichSpout;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;


public class WordSpout extends baseRichSpout {

    private SpoutOutputCollector collector;
    private static String[] words = {"星期一","星期二","星期三","星期四","星期五","星期六","星期日"};

    public WordSpout() {
        System.out.println("--====================WordSpout===---------------");
    }

    public void nextTuple() {
        //随机取 words 字符串中一个词。
        String word = words[new Random().nextInt(words.length)];
        //发射元组到输出收集器
        collector.emit(new Values(word));

    }

    public void open(Map arg0, TopologyContext arg1, SpoutOutputCollector arg2) {
        this.collector=arg2;
        //定义数据源输出收集器
    }

    public void declareOutputFields(OutputFieldsDeclarer arg0) {
        // TODO Auto-generated method stub
        //声明输出字段的名称为为 word
        arg0.declare(new Fields("word"));
    }

}

(2)整理数据流并输出

package test.storm;


import java.util.Map;

import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.baseRichBolt;
import org.apache.storm.tuple.Tuple;

public class ProcessBolt extends baseRichBolt {

    public ProcessBolt() {
        // TODO Auto-generated constructor stub
        System.out.println("-----------ProcessBolt==============");
    }

    @Override
    public void execute(Tuple arg0) {
        // TODO Auto-generated method stub
        //此处直接对接受到的元组进行处理,然后输出到控制台,这里没有将处理后的数据再送到输出收集器中。
        //取得元组的数据
        String word = (String) arg0.getValue(0);
        String out = "Hello :" + word + "!";
        //输出到控制台,使用 err.println 会显示红色,所以这里使用 err
        System.err.println(out);
    }
    @Override
    public void prepare(Map arg0, TopologyContext arg1, OutputCollector arg2) {
        // TODO Auto-generated method stub
        System.out.println("-----------prepare==============");

    }
    @Override
    public void declareOutputFields(OutputFieldsDeclarer arg0) {
        // TODO Auto-generated method stub
        System.out.println("-----------declareOutputFields==============");


    }

}
 

(3)编写启动类:

package test.storm;



import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.topology.TopologyBuilder;



public class WordStorm {

    public WordStorm() {
        // TODO Auto-generated constructor stub
    }

    public static void main(String[] args)  throws Exception {
        // TODO Auto-generated method stub
        //定义 TopologyBuilder
        TopologyBuilder builder=new TopologyBuilder();
        //定义 Spout
        builder.setSpout("Spout_ID", new WordSpout());
        //定义 Bolt
        builder.setBolt("Bolt_ID", new ProcessBolt()).localOrShuffleGrouping("Spout_ID");

        //下面开始定义运行模式
        final Config config=new Config();
        config.setDebug(true);
        //设置workers
        config.setNumWorkers(1);
        config.setMaxSpoutPending(1);

        if (args != null && args.length > 0) {
            //集群运行模式
            config.setNumWorkers(3);

            StormSubmitter.submitTopologyWithProgressBar(args[0], config, builder.createTopology());
        }else {
            //使用本地模式运行
            final LocalCluster localCluster=new LocalCluster();

            localCluster.submitTopology(WordStorm.class.getSimpleName(), config, builder.createTopology());
            org.apache.storm.utils.Utils.sleep(20000);

            localCluster.killTopology(WordStorm.class.getSimpleName());
            localCluster.shutdown();
        }


    }

}

运行结果:

 

欢迎分享,转载请注明来源:内存溢出

原文地址: http://outofmemory.cn/zaji/5634835.html

(0)
打赏 微信扫一扫 微信扫一扫 支付宝扫一扫 支付宝扫一扫
上一篇 2022-12-16
下一篇 2022-12-15

发表评论

登录后才能评论

评论列表(0条)

保存