spout介紹

一個spout是由流組成的資料來源在storm的拓撲裡，通常情況下會讀取外部的資料來源 
然後emit（發射）到拓撲裡面，比如是kafka，MySQL或者redis等等，Spout有兩種實現一種是可靠的訊息實現，如果傳送失敗則會重試，另外一種是不可靠的訊息實現可能會出現訊息丟失，spout可以一次宣告多個數據流通過OutputFieldsDeclarer類的declareStream方法，當然前提是你的SpoutOutputCollector裡的emit也是多個流 

Spout裡面主要的方法是nextTuple，它裡面可以發射新的tuple到拓撲，或者當沒有訊息的時候就return，需要注意，這個方法裡面不能阻塞，因為storm呼叫spout方法是單執行緒的，其他的主要方法是ack和fail，如果使用了可靠的spout，可以使用ack和fail來確定訊息傳送狀態 

相關擴充套件： 
IRichSpout：spout類必須實現的介面 
BaseRichSpout ：可靠的spout有ack確保 
BaseBasicSpout ：不可靠的spout 

1.Spout元件：建立Spout（WordCountSpout）元件採集資料，作為整個Topology的資料來源

WordCountSpout.java

package storm;

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;
import org.apache.storm.utils.Utils;
import java.util.Map;
import java.util.Random;

public class WordCountSpout extends BaseRichSpout {

    private SpoutOutputCollector collector;
    //模擬產生一些資料
    private String[] data = {"I I love Beijing","I love love love  China","Beijing is id is is the the capital of China"};

    /**
     * open方法的作用主要是將collector進行初始化
     * collector的作用：將採集到的資料傳送給下一個元件
     */
    @Override
    public void open(Map map, TopologyContext topologyContext, SpoutOutputCollector collector) {
        this.collector=collector;
    }

    @Override
    public void nextTuple() {
        Utils.sleep(3000);
        int random = (new Random()).nextInt(3);
        String value = data[random];
        System.out.println("產生的隨機值是"+value);
        //傳送給下一個元件
        collector.emit(new Values(value));
    }


   //申明發送給下一個元件的tuple的schema（結構）
    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("sentence"));
    }
}
 

bolt介紹

Bolts 業務處理單元 
所有的拓撲處理都會在bolt中進行，bolt裡面可以做任何etl，比如過濾，函式，聚合，連線，寫入資料庫系統或快取等，一個bolt可以做簡單的事件流轉換，如果是複雜的流轉化，往往需要多個bolt參與，這就是流計算，每個bolt都進行一個業務邏輯處理，bolt也可以emit多個流到下游，通過declareStream方法宣告輸出的schema。 

Bolt裡面主要的方法是execute方法，每次處理一個輸入的tuple，bolt裡面也可以發射新的tuple使用OutputCollector類，bolt裡面每處理一個tuple必須呼叫ack方法以便於storm知道某個tuple何時處理完成。Strom裡面的IBasicBolt介面可以自動 
呼叫ack。 

相關拓展： 
IRichBolt：bolts的通用介面 
IBasicBolt：擴充套件的bolt介面，可以自動處理ack 
OutputCollector：bolt發射tuple到下游bolt裡面 

2.Bolt元件1：建立Bolt（WordCountSplitBolt）元件進行分詞操作

WordCountSplitBolt.java

package storm;

import com.google.common.collect.Maps;
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.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import java.util.Map;

public class WordCountSplitBolt extends BaseRichBolt{

    //bolt元件的收集器 用於將資料傳送給下一個bolt
    private OutputCollector collector;


    //初始化
    @Override
    public void prepare(Map map, TopologyContext topologyContext, OutputCollector collector) {
        this.collector = collector;
    }

    @Override
    public void execute(Tuple tuple) {
        //處理上一級發來的資料
        String value = tuple.getStringByField("sentence");
        String[] data= value.split(" ");
        //輸出
        for (String word : data){
            collector.emit(new Values(word,1));
        }
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        //申明發送給下一個元件的tuple schema結構
        declarer.declare(new Fields("word","count"));
    }
}

3.Bolt元件2：建立Bolt（WordCountTotalBolt）元件進行單詞統計操作

WordCountTotalBolt.java

package storm;

import com.google.common.collect.Maps;
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.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import java.util.Map;
import java.util.Set;

public class WordCountTotalBolt extends BaseRichBolt{

    private OutputCollector collector;
    Map<String,Integer> result=Maps.newHashMap();
    @Override
    public void prepare(Map map, TopologyContext topologyContext, OutputCollector collector) {
        this.collector = collector;
    }

    @Override
    public void execute(Tuple tuple) {
        String word = tuple.getStringByField("word");
        Integer count = tuple.getIntegerByField("count");

        if (result.get(word) == null){
            result.put(word,count);
        }else {
            result.put(word,count + result.get(word));
        }
        result.entrySet().forEach(enty-> System.out.println("單詞："+enty.getKey()+" " + "數量："+ enty.getValue()));

        collector.emit(new Values(word,result.get(word)));

    }

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

4.Topology主程式：（WordCountTopology）

WordCountTopology.java

package storm;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.generated.StormTopology;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.tuple.Fields;

public class WordCountTopology {
    public static void main(String[] args) {
        TopologyBuilder builder = new TopologyBuilder();

        //1 指定任務的spout元件
        builder.setSpout("1",new WordCountSpout());

        //2 指定任務的第一個bolt元件
        builder.setBolt("2",new WordCountSplitBolt()).shuffleGrouping("1");

        //3 指定任務的第二個bolt元件
        builder.setBolt("3",new WordCountTotalBolt()).fieldsGrouping("2",new Fields("word"));

        //建立任務
        StormTopology job = builder.createTopology();

        Config config = new Config();

        //執行任務有兩種模式
        //1 本地模式   2 叢集模式

        //1、本地模式
        LocalCluster localCluster = new LocalCluster();
        localCluster.submitTopology("MyWordCount",config,job);

        //2、叢集模式：用於打包jar，並放到storm執行
//        StormSubmitter.submitTopology(args[0], conf, job);
    }
}

pom.xml

<dependency>
<groupId>org.apache.storm</groupId>
<artifactId>storm-core</artifactId>
<version>1.0.3</version>
<!--<scope>provided</scope>-->
</dependency>

<dependency>
<groupId>org.apache.storm</groupId>
<artifactId>storm-rename-hack</artifactId>
<version>1.0.3</version>
</dependency>

<dependency>
<groupId>org.apache.storm</groupId>
<artifactId>storm-hbase</artifactId>
<version>1.0.3</version>
<scope>test</scope>
</dependency>

<dependency>
<groupId>org.apache.storm</groupId>
<artifactId>storm-redis</artifactId>
<version>1.0.3</version>
</dependency>