流式計算--storm3(Storm單詞技術案例)

阿新 • • 發佈：2018-12-22

功能說明:設計一個topology，來實現對文件裡面的單詞出現的頻率進行統計。本篇部落格是在storm概念講解和storm叢集搭建的基礎上來的

1.建立一個maven專案：

新增以來如下：

 <dependency>
                <groupId>org.apache.storm</groupId>
                <artifactId>storm-core</artifactId>
                <version>1.0.6</version>
                <!--storm叢集裡面已經有jar包，所以新增下面的provided表示已提供jar包，所以打包的時候就要剔除相應的jar包，現在跑本地需要註釋掉-->
                <!--<scope>provided</scope>-->
            </dependency>

建立一個MySpout類繼承BaseRichSpout類，用於接收外部資料來源的資料

package com.wx.storm1;

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 java.util.Map;

public class MySpout extends BaseRichSpout {

    SpoutOutputCollector outputCollector;
    //初始化的方法,該方法呼叫一次，主要由Storm框架傳入SpoutOutputCollector
    public void open(Map map, TopologyContext topologyContext, SpoutOutputCollector spoutOutputCollector) {
             this.outputCollector=spoutOutputCollector;
    }
    //storm 框架在 while(true) 呼叫nextTuple方法
    public void nextTuple() {
        outputCollector.emit(new Values("I love you,wang zu xian,my names Mr.Wang"));
    }
    //訊息源可以傳送多條訊息流stream,多條訊息流可以理解為多種型別的資料
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        outputFieldsDeclarer.declare(new Fields("word"));
    }
}

建立一個MyBolt繼承於BaseRichBolt，主要是分割單詞，得到的單詞計數輸出

package com.wx.storm1;

import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.IBasicBolt;
import org.apache.storm.topology.IRichBolt;
import org.apache.storm.topology.IWindowedBolt;
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 MyBolt extends BaseRichBolt{
    OutputCollector outputCollector;
    //初始化方法
    public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {
        this.outputCollector=outputCollector;
    }
    //被storm框架 while(true) 迴圈呼叫  傳入引數tuple
    public void execute(Tuple tuple) {
        //這裡將單詞分割，得到的單詞計數輸出
        String string = tuple.getString(0);
        String[] split = string.split(" ");
        for (String val:split)
        {
            outputCollector.emit(new Values(val,1));
        }
    }
    //輸出到下一個bolt
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        outputFieldsDeclarer.declare(new Fields("word","num"));
    }
}

建立一個MyBoltCount繼承於BaseRichBolt ，主要是對相同的單詞做統計

package com.wx.storm1;

import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.IBasicBolt;
import org.apache.storm.topology.IRichBolt;
import org.apache.storm.topology.IWindowedBolt;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Tuple;

import java.util.HashMap;
import java.util.Map;

public class MyBoltCount extends BaseRichBolt {
    OutputCollector outputCollector;
    Map<String,Integer> map=new HashMap<String, Integer>();
    //初始化方法
    public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {
        this.outputCollector=outputCollector;
    }
    //被storm框架 while(true) 迴圈呼叫  傳入引數tuple
    public void execute(Tuple tuple) {
          //這裡就將相同的單詞統計出來，然後輸出
        String word = tuple.getString(0);
        Integer num = tuple.getInteger(1);
        if(!map.containsKey(word))
        {
            map.put(word,num);
        }
        else
        {
            Integer count=map.get(word);
            map.put(word,num+count);
        }
        System.out.println("count:"+map);
    }

    //這個bolt就是盡頭了，不輸出
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {

    }
}

建立一個拓撲來分配任務：

package com.wx.storm1;
import org.apache.storm.Config;
import org.apache.storm.StormSubmitter;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.tuple.Fields;

public class WordCountStrom {
     public static  void  main(String[] args) throws  Exception
     {
         //Storm計算wordcount
         //1.建立一個TopologyBuilder
         TopologyBuilder topologyBuilder = new TopologyBuilder();
         //指定Spout的類和需要task的數量，task的數量即執行緒的數量
         topologyBuilder.setSpout("myspout",new MySpout(),2);
         //指定需要的第一級bolt的類和task的數量
         topologyBuilder.setBolt("mybolt1",new MyBolt(),2).shuffleGrouping("myspout"); //隨機函式分組
         topologyBuilder.setBolt("mybolt2",new MyBoltCount(),4).fieldsGrouping("mybolt1",new Fields("word"));//hash取模分組

         //2.建立一個configuration，用來指定當前topology 需要的worker的數量
         Config config=new Config();
         config.setNumWorkers(3);

         //3.提交任務  -----兩種模式 本地模式和叢集模式
         //StormSubmitter.submitTopology("wordcount",config,topologyBuilder.createTopology());
         LocalCluster localCluster = new LocalCluster();
         localCluster.submitTopology("wordcount",config,topologyBuilder.createTopology());

     }
}

測試結果：

如果要提交到叢集執行：
首先需要修改pom檔案，表示storm叢集中已經提供jar包

其次提交拓撲的任務修改為叢集模式

在idea中進行打包：雙擊package打包成功：

打的jar包在target目錄下：ok ,上傳至叢集，準備執行

執行執行

叢集的拓撲途中就有了：

使用命令把他kill 掉：storm kill wordcount

可以檢視日誌：cd /export/servers/storm/logs/

2.Storm常用操作命令

有許多簡單且有用的命令可以用來管理拓撲，它們可以提交、殺死、禁用、再平衡拓撲。

1.提交任務命令格式：storm jar 【jar路徑】【拓撲包名.拓撲類名】【拓撲名稱】

bin/storm jar examples/storm-starter/storm-starter-topologies-0.9.6.jar storm.starter.WordCountTopology wordcount

2.殺死任務命令格式：storm kill 【拓撲名稱】 -w 10（執行kill命令時可以通過-w [等待秒數]指定拓撲停用以後的等待時間）

storm kill topology-name -w 10

3.停用任務命令格式：storm deactivte 【拓撲名稱】

storm deactivte topology-name

我們能夠掛起或停用執行中的拓撲。當停用拓撲時，所有已分發的元組都會得到處理，但是spouts的nextTuple方法不會被呼叫。銷燬一個拓撲，可以使用kill命令。它會以一種安全的方式銷燬一個拓撲，首先停用拓撲，在等待拓撲訊息的時間段內允許拓撲完成當前的資料流。

4.啟用任務命令格式：storm activate【拓撲名稱】

storm activate topology-name

5.重新部署任務命令格式：storm rebalance 【拓撲名稱】

storm rebalance topology-name

再平衡使你重分配叢集任務。這是個很強大的命令。比如，你向一個執行中的叢集增加了節點。再平衡命令將會停用拓撲，然後在相應超時時間之後重分配工人，並重啟拓撲。

3.執行原理圖（Mr.Mao的圖）

這裡我們stream grouping的分組方式是按欄位進行分組，比如按userid來分組，具有同樣userid的tuple會被分到相同的Bolts裡的一個task，而不同的userid則會被分配到不同的bolts裡的task。

4.Stream Grouping詳解

Storm裡面有7種類型的stream grouping

Shuffle Grouping: 隨機分組，隨機派發stream裡面的tuple，保證每個bolt接收到的tuple數目大致相同。
Fields Grouping：按欄位分組，比如按userid來分組，具有同樣userid的tuple會被分到相同的Bolts裡的一個task，而不同的userid則會被分配到不同的bolts裡的task。
All Grouping：廣播發送，對於每一個tuple，所有的bolts都會收到。
Global Grouping：全域性分組，這個tuple被分配到storm中的一個bolt的其中一個task。再具體一點就是分配給id值最低的那個task。
Non Grouping：不分組，這stream grouping個分組的意思是說stream不關心到底誰會收到它的tuple。目前這種分組和Shuffle grouping是一樣的效果， 有一點不同的是storm會把這個bolt放到這個bolt的訂閱者同一個執行緒裡面去執行。
Direct Grouping：直接分組，這是一種比較特別的分組方法，用這種分組意味著訊息的傳送者指定由訊息接收者的哪個task處理這個訊息。只有被宣告為Direct Stream的訊息流可以宣告這種分組方法。而且這種訊息tuple必須使用emitDirect方法來發射。訊息處理者可以通過TopologyContext來獲取處理它的訊息的task的id （OutputCollector.emit方法也會返回task的id）。
Local or shuffle grouping：如果目標bolt有一個或者多個task在同一個工作程序中，tuple將會被隨機發生給這些tasks。否則，和普通的Shuffle Grouping行為一致。

5.問題（Mr.Mao的圖）

1.叢集如何啟動，任務如何執行？

2、叢集如何通訊：叢集架構中的各個模組是如何通訊的？拓撲程式中的各個Task是如何通訊的？

nimbus與supervisor之間是通過zk進行通訊的

不同機器上的worker是基於網路io的socket通訊（netty）

worker內部是基於縣城的通訊，執行緒於執行緒通訊

3.如何保證訊息的不丟失 ack-fail機制如何實現的？

4.storm 啟動流程分析

流式計算一般架構圖

下面會學習kafka....

流式計算--storm3(Storm單詞技術案例)

1.建立一個maven專案：

2.Storm常用操作命令

3.執行原理圖（Mr.Mao的圖）

4.Stream Grouping詳解

5.問題（Mr.Mao的圖）

流式計算一般架構圖

流式計算--storm3(Storm單詞技術案例)

流式計算--storm1(storm概念初識)

Storm單詞技術案例

分散式流式計算框架 Storm

流式計算之Storm簡介

storm 流式計算框架

Storm簡介——實時流式計算介紹

hadoop（十三）storm流式計算（實時處理）

大資料學習：storm流式計算

流式計算--整合kafka+flume+storm

【流式計算】Twitter Storm原始碼分析之Nimbus/Supervisor本地目錄結構

流式計算storm應用場景簡介

Storm流式計算入門

【流式計算】Twitter Storm原始碼分析之ZooKeeper中的目錄結構

大資料流式計算：關鍵技術及系統例項

大資料流式計算 storm 基礎

Storm流式計算小結+叢集故障恢復

流式處理框架storm淺析

流式計算簡介

Flink 流式計算框架（學習一）

流式計算--storm3(Storm單詞技術案例)

1.建立一個maven專案：

2.Storm常用操作命令

3.執行原理圖（Mr.Mao的圖）

4.Stream Grouping詳解

5.問題（Mr.Mao的圖）

流式計算一般架構圖

相關推薦