jstorm系列-1:入门

Posted

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了jstorm系列-1:入门相关的知识,希望对你有一定的参考价值。

有了基本的概念之后,我们用jstorm来做一点小事情吧

做一个很无聊的事情:给定一个时间戳,输出对应的问候语

规则是:时间戳的十位对应的数字对应不同的时间段,0-2代表早上,3代表中午,4-6代表下午,7-9代表晚上,分别输出早上,中午,下午和晚上。

我们用spout来发送时间戳,bolt来处理时间戳并输出对应的问候语,并且统计每一时间段的问候数目,判断时间戳的生成是否随机。

代码如下:

public class TimeStampSpout implements IRichSpout{
    private static final long serialVersionUID = 1L;
    private SpoutOutputCollector collector;

    @Override
    。。。。一切没有用到的函数不展示

    @Override
    public void nextTuple() {
        long now = System.currentTimeMillis();
        Values tuple = new Values(now);
        System.out.println("spout:"+tuple);
        this.collector.emit(tuple);
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
        this.collector = collector;
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("timestamp"));
    }

}

 

public class GreetingBolt implements IRichBolt{
    private static final long serialVersionUID = 1L;
    private static Map<String, Integer> count = new TreeMap<String, Integer>(); 
    private final String morning = "morning"; 
    private final String noon = "noon"; 
    private final String afternoon = "afternoon"; 
    private final String evening = "evening";
    private static Integer total = 0;
    @Override
    public void cleanup() {
    }

    @Override
    public void execute(Tuple input) {
        long timestamp= input.getLong(0);
//        System.out.println("bolt:"+timestamp);
        total+=1;
//        获取十位数
        long second = (timestamp/10)%10;
        if(second<3){
            System.out.println("bolt:"+morning); 
            count.put(morning, (count.get(morning)==null)?1:count.get(morning)+1);
        }
        else if(second==3){
            System.out.println("bolt:"+noon);
            count.put(noon, (count.get(noon)==null)?1:count.get(noon)+1);
        }
        else if(second<8){
            System.out.println("bolt:"+afternoon);
            count.put(afternoon, (count.get(afternoon)==null)?1:count.get(afternoon)+1);
        }
        else{
            System.out.println("bolt:"+evening);
            count.put(evening, (count.get(evening)==null?1:count.get(evening)+1));
        }
        if(total%10==0){
            System.out.println("distribution show as followed:");
            System.out.println(morning+":"+1.0*((count.get(morning)==null)?0:count.get(morning))/total);
            System.out.println(noon+":"+1.0*((count.get(noon)==null?0:count.get(noon)))/total);
            System.out.println(afternoon+":"+1.0*((count.get(afternoon)==null?0:count.get(afternoon)))/total);
            System.out.println(evening+":"+1.0*((count.get(evening)==null?0:count.get(evening)))/total);
        }
    }

}

 

 

public class TestTopology {
    public static void main(String[] args) {
        TopologyBuilder builder = new TopologyBuilder();
        builder.setSpout("TimeStampSpout", new TimeStampSpout());
        builder.setBolt("GreetingBolt",new GreetingBolt()).shuffleGrouping("TimeStampSpout");
        LocalCluster cluster = new LocalCluster();
        Config config = new Config();
        cluster.submitTopology("test", config, builder.createTopology());
    }
}

 

 

p

以上是关于jstorm系列-1:入门的主要内容,如果未能解决你的问题,请参考以下文章

jstorm 核心

JStorm之Topology调度

JStorm:概念与编程模型

[转]JStorm之Nimbus简介

jstorm之于storm

JStorm第一个程序WordCount详解