Storm详解二、写第一个Storm应用

栏目: Java · 发布时间: 7年前

内容简介:Storm详解二、写第一个Storm应用

在全面介绍Storm之前,我们先通过一个简单的Demo让大家整体感受一下什么是Storm。

Storm运行模式:

  1. 本地模式(Local Mode): 即Topology(相当于一个任务,后续会详细讲解)  运行在本地机器的单一JVM上,这个模式主要用来开发、调试。
  2. 远程模式(Remote Mode):在这个模式,我们把我们的Topology提交到集群,在这个模式中,Storm的所有组件都是线程安全的,因为它们都会运行在不同的Jvm或物理机器上,这个模式就是正式的生产模式。

写一个HelloWord Storm

  我们现在创建这么一个应用,统计文本文件中的单词个数,详细学习过Hadoop的朋友都应该写过。那么我们需要具体创建这样一个Topology,用一个spout负责读取文本文件,用第一个bolt来解析成单词,用第二个bolt来对解析出的单词计数,整体结构如图所示:

可以从这里下载源码: https://github.com/storm-book/examples-ch02-getting_started/zipball/master

Storm详解二、写第一个Storm应用

  写一个可运行的Demo很简单,我们只需要三步:

  1. 创建一个Spout读取数据
  2. 创建bolt处理数据
  3. 创建一个Topology提交到集群

下面我们就写一下,以下代码拷贝到eclipse(依赖的jar包到官网下载即可)即可运行。

1.创建一个Spout作为数据源

Spout作为数据源,它实现了IRichSpout接口,功能是读取一个文本文件并把它的每一行内容发送给bolt。

package storm.demo.spout;  
  
import java.io.BufferedReader;  
import java.io.FileNotFoundException;  
import java.io.FileReader;  
import java.util.Map;  
import backtype.storm.spout.SpoutOutputCollector;  
import backtype.storm.task.TopologyContext;  
import backtype.storm.topology.IRichSpout;  
import backtype.storm.topology.OutputFieldsDeclarer;  
import backtype.storm.tuple.Fields;  
import backtype.storm.tuple.Values;  
public class WordReader implements IRichSpout {  
    private static final long serialVersionUID = 1L;  
    private SpoutOutputCollector collector;  
    private FileReader fileReader;  
    private boolean completed = false;  
  
    public boolean isDistributed() {  
        return false;  
    }  
    /** 
     * 这是第一个方法,里面接收了三个参数,第一个是创建Topology时的配置, 
     * 第二个是所有的Topology数据,第三个是用来把Spout的数据发射给bolt 
     * **/  
    @Override  
    public void open(Map conf, TopologyContext context,  
            SpoutOutputCollector collector) {  
        try {  
            //获取创建Topology时指定的要读取的文件路径  
            this.fileReader = new FileReader(conf.get("wordsFile").toString());  
        } catch (FileNotFoundException e) {  
            throw new RuntimeException("Error reading file ["  
                    + conf.get("wordFile") + "]");  
        }  
        //初始化发射器  
        this.collector = collector;  
  
    }  
    /** 
     * 这是Spout最主要的方法,在这里我们读取文本文件,并把它的每一行发射出去(给bolt) 
     * 这个方法会不断被调用,为了降低它对CPU的消耗,当任务完成时让它sleep一下 
     * **/  
    @Override  
    public void nextTuple() {  
        if (completed) {  
            try {  
                Thread.sleep(1000);  
            } catch (InterruptedException e) {  
                // Do nothing  
            }  
            return;  
        }  
        String str;  
        // Open the reader  
        BufferedReader reader = new BufferedReader(fileReader);  
        try {  
            // Read all lines  
            while ((str = reader.readLine()) != null) {  
                /** 
                 * 发射每一行,Values是一个ArrayList的实现 
                 */  
                this.collector.emit(new Values(str), str);  
            }  
        } catch (Exception e) {  
            throw new RuntimeException("Error reading tuple", e);  
        } finally {  
            completed = true;  
        }  
  
    }  
    @Override  
    public void declareOutputFields(OutputFieldsDeclarer declarer) {  
        declarer.declare(new Fields("line"));  
  
    }  
    @Override  
    public void close() {  
        // TODO Auto-generated method stub  
    }  
      
    @Override  
    public void activate() {  
        // TODO Auto-generated method stub  
  
    }  
    @Override  
    public void deactivate() {  
        // TODO Auto-generated method stub  
  
    }  
    @Override  
    public void ack(Object msgId) {  
        System.out.println("OK:" + msgId);  
    }  
    @Override  
    public void fail(Object msgId) {  
        System.out.println("FAIL:" + msgId);  
  
    }  
    @Override  
    public Map<String, Object> getComponentConfiguration() {  
        // TODO Auto-generated method stub  
        return null;  
    }  
}  

2.创建两个bolt来处理Spout发射出的数据

Spout已经成功读取文件并把每一行作为一个tuple(在Storm数据以tuple的形式传递)发射过来,我们这里需要创建两个bolt分别来负责解析每一行和对单词计数。

Bolt中最重要的是execute方法,每当一个tuple传过来时它便会被调用。

第一个bolt:WordNormalizer

package storm.demo.bolt;  
import java.util.ArrayList;  
import java.util.List;  
import java.util.Map;  
import backtype.storm.task.OutputCollector;  
import backtype.storm.task.TopologyContext;  
import backtype.storm.topology.IRichBolt;  
import backtype.storm.topology.OutputFieldsDeclarer;  
import backtype.storm.tuple.Fields;  
import backtype.storm.tuple.Tuple;  
import backtype.storm.tuple.Values;  
public class WordNormalizer implements IRichBolt {  
    private OutputCollector collector;  
    @Override  
    public void prepare(Map stormConf, TopologyContext context,  
            OutputCollector collector) {  
        this.collector = collector;  
    }  
    /**这是bolt中最重要的方法,每当接收到一个tuple时,此方法便被调用 
     * 这个方法的作用就是把文本文件中的每一行切分成一个个单词,并把这些单词发射出去(给下一个bolt处理) 
     * **/  
    @Override  
    public void execute(Tuple input) {  
        String sentence = input.getString(0);  
        String[] words = sentence.split(" ");  
        for (String word : words) {  
            word = word.trim();  
            if (!word.isEmpty()) {  
                word = word.toLowerCase();  
                // Emit the word  
                List a = new ArrayList();  
                a.add(input);  
                collector.emit(a, new Values(word));  
            }  
        }  
        //确认成功处理一个tuple  
        collector.ack(input);  
    }  
    @Override  
    public void declareOutputFields(OutputFieldsDeclarer declarer) {  
        declarer.declare(new Fields("word"));  
  
    }  
    @Override  
    public void cleanup() {  
        // TODO Auto-generated method stub  
  
    }  
    @Override  
    public Map<String, Object> getComponentConfiguration() {  
        // TODO Auto-generated method stub  
        return null;  
    }  
}  

第二个bolt:WordCounter

package storm.demo.bolt;  
import java.util.HashMap;  
import java.util.Map;  
import backtype.storm.task.OutputCollector;  
import backtype.storm.task.TopologyContext;  
import backtype.storm.topology.IRichBolt;  
import backtype.storm.topology.OutputFieldsDeclarer;  
import backtype.storm.tuple.Tuple;  
  
public class WordCounter implements IRichBolt {  
    Integer id;  
    String name;  
    Map<String, Integer> counters;  
    private OutputCollector collector;  
  
    @Override  
    public void prepare(Map stormConf, TopologyContext context,  
            OutputCollector collector) {  
        this.counters = new HashMap<String, Integer>();  
        this.collector = collector;  
        this.name = context.getThisComponentId();  
        this.id = context.getThisTaskId();  
  
    }  
    @Override  
    public void execute(Tuple input) {  
        String str = input.getString(0);  
        if (!counters.containsKey(str)) {  
            counters.put(str, 1);  
        } else {  
            Integer c = counters.get(str) + 1;  
            counters.put(str, c);  
        }  
        // 确认成功处理一个tuple  
        collector.ack(input);  
    }  
    /** 
     * Topology执行完毕的清理工作,比如关闭连接、释放资源等操作都会写在这里 
     * 因为这只是个Demo,我们用它来打印我们的计数器 
     * */  
    @Override  
    public void cleanup() {  
        System.out.println("-- Word Counter [" + name + "-" + id + "] --");  
        for (Map.Entry<String, Integer> entry : counters.entrySet()) {  
            System.out.println(entry.getKey() + ": " + entry.getValue());  
        }  
        counters.clear();  
    }  
    @Override  
    public void declareOutputFields(OutputFieldsDeclarer declarer) {  
        // TODO Auto-generated method stub  
  
    }  
    @Override  
    public Map<String, Object> getComponentConfiguration() {  
        // TODO Auto-generated method stub  
        return null;  
    }  
}  

3.在main函数中创建一个Topology

在这里我们要创建一个Topology和一个LocalCluster对象,还有一个Config对象做一些配置。   

package storm.demo;  
  
import storm.demo.bolt.WordCounter;  
import storm.demo.bolt.WordNormalizer;  
import storm.demo.spout.WordReader;  
import backtype.storm.Config;  
import backtype.storm.LocalCluster;  
import backtype.storm.topology.TopologyBuilder;  
import backtype.storm.tuple.Fields;  
public class WordCountTopologyMain {  
    public static void main(String[] args) throws InterruptedException {  
        //定义一个Topology  
        TopologyBuilder builder = new TopologyBuilder();  
        builder.setSpout("word-reader",new WordReader());  
        builder.setBolt("word-normalizer", new WordNormalizer())  
        .shuffleGrouping("word-reader");  
        builder.setBolt("word-counter", new WordCounter(),2)  
        .fieldsGrouping("word-normalizer", new Fields("word"));  
        //配置  
        Config conf = new Config();  
        conf.put("wordsFile", "d:/text.txt");  
        conf.setDebug(false);  
        //提交Topology  
        conf.put(Config.TOPOLOGY_MAX_SPOUT_PENDING, 1);  
        //创建一个本地模式cluster  
        LocalCluster cluster = new LocalCluster();  
        cluster.submitTopology("Getting-Started-Toplogie", conf,  
        builder.createTopology());  
        Thread.sleep(1000);  
        cluster.shutdown();  
    }  
}  

运行这个函数我们即可看到后台打印出来的单词个数。

(ps:因为是Local模式,运行开始可能会打印很多错误log,这个先不用管)

Storm详解二、写第一个Storm应用


以上所述就是小编给大家介绍的《Storm详解二、写第一个Storm应用》,希望对大家有所帮助,如果大家有任何疑问请给我留言,小编会及时回复大家的。在此也非常感谢大家对 码农网 的支持!

查看所有标签

猜你喜欢:

本站部分资源来源于网络,本站转载出于传递更多信息之目的,版权归原作者或者来源机构所有,如转载稿涉及版权问题,请联系我们

解决网页设计一定会遇到的210个问题

解决网页设计一定会遇到的210个问题

2006-4 / 42.00元

如何选择适合、简单、方便、快速的方法来解决您的网页设计问题?不会HTML、JavaScript、CSS也可轻易完成许多网页功能与特效。本书包含上百种HTML、JavaScript、CSS使用应用技巧与盲点解说,包含10个常用表单资料判断函数与特殊技巧,不必修改就可用于任何网页。本书现有的多数网页设计书籍相辅相成,让您事半功倍地完成工作。   许多计算机书籍都是从某个语言或者某个软件的......一起来看看 《解决网页设计一定会遇到的210个问题》 这本书的介绍吧!

HEX HSV 转换工具
HEX HSV 转换工具

HEX HSV 互换工具

HSV CMYK 转换工具
HSV CMYK 转换工具

HSV CMYK互换工具