一、編寫wordcount案例

1、新建java項目

2、添加storm的jar包

storm軟件包中l(wèi)ib目錄下的所有jar包

3、編寫java類

WordCountTopology.java

package com.bjsxt.storm.wc;

import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.generated.StormTopology;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.tuple.Fields;
import backtype.storm.utils.Utils;

public class WordCountTopology {

    public static void main(String[] args) {
        // 拓撲封裝了計算邏輯
        TopologyBuilder builder = new TopologyBuilder();
        // 設置lineSpout：該spout負責向拓撲發(fā)送句子
        builder.setSpout("lineSpout", new LineSpout());
        // 設置切分閃電，該閃電處理從水龍頭lineSpout通過隨機分組發(fā)送過來的元組
        builder.setBolt("splitBolt", new SplitBolt())
            .shuffleGrouping("lineSpout");
        // 定義一個計數(shù)閃電，該閃電從splitBolt閃電通過按字段分組的方式分發(fā)過來的元組
        // 按照元組中word的值進行分組。要保證相同的單詞一定發(fā)送給同一個閃電。
        builder.setBolt("countBolt", new CountBolt())
                .fieldsGrouping("splitBolt", new Fields("word"));

        // 通過建造者創(chuàng)建一個拓撲的實例
        StormTopology wordCountTopology = builder.createTopology();

        // 本地模擬集群
        LocalCluster cluster = new LocalCluster();

        Config config = new Config();

        // 將拓撲提交到本地模擬集群
        cluster.submitTopology("wordCountTopology", config, wordCountTopology);

        // 睡眠10s，也就是讓本地模擬集群運行10s
        Utils.sleep(10000);

        // 關閉本地模擬集群
        cluster.shutdown();

    }

}

LineSpout.java

package com.bjsxt.storm.wc;

import backtype.storm.spout.SpoutOutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichSpout;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;
import backtype.storm.utils.Utils;

import java.util.Map;

public class LineSpout extends BaseRichSpout {

    private SpoutOutputCollector collector;

    private String[] lines = {
            "The logic for a realtime application is packaged into a Storm topology",
            "A stream is an unbounded sequence of tuples that is processed and created in parallel in a distributed fashion",
            "A spout is a source of streams in a topology",
            "Bolts can do anything from filtering, functions, aggregations, joins, talking to databases, and more.",
            "A stream grouping defines how that stream should be partitioned among the bolt's tasks.",
            "Storm guarantees that every spout tuple will be fully processed by the topology",
            "Each spout or bolt executes as many tasks across the cluster",
            "Each worker process is a physical JVM and executes a subset of all the tasks for the topology"
    };

    private int index = 0;

    @Override
    public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
        // 在該組件在集群中初始化的時候調用一次
        this.collector = collector;
    }

    @Override
    public void nextTuple() {
        // 由storm的線程不停地調用，以便從數(shù)據(jù)源獲取元組
        // 該方法不需要自己寫循環(huán)和遍歷
        // 該方法不能阻塞
        // 負責從數(shù)據(jù)源獲取元組，向DAG發(fā)送元組
        // 輪詢取出句子
        String lingStr = lines[index % lines.length];
        // 將句子封裝為元組發(fā)射
        collector.emit(new Values(lingStr));
        index++;

        Utils.sleep(10);
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        // 用于聲明元組的結構以及流
//        declarer.declareStream("s1", new Fields("key1", "key2", "key3"));
//        declarer.declareStream("s2", new Fields("key21", "key22"));
        // 發(fā)送元組的時候就有一個字段，是line，它的值是句子
        // 可以將元組想象為map集合，只不過其key是固定的幾個
        declarer.declare(new Fields("line"));
    }
}

SplitBolt.java

package com.bjsxt.storm.wc;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;

import java.util.Map;

public class SplitBolt extends BaseRichBolt {

    private OutputCollector collector;

    @Override
    public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
        this.collector = collector;
    }

    @Override
    public void execute(Tuple input) {
        String lineStr = input.getStringByField("line");
        String[] wordStrs = lineStr.split(" ");

        for (String wordStr : wordStrs) {
            // <hello, 1>
            this.collector.emit(new Values(wordStr, 1));
        }

    }

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

CountBolt.java

package com.bjsxt.storm.wc;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Tuple;

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

public class CountBolt extends BaseRichBolt {

    private Map<String, Integer> counts;

    @Override
    public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
        counts = new HashMap<>();
    }

    @Override
    public void execute(Tuple input) {
//        new Fields("word", "count")
        String wordStr = input.getStringByField("word");
        Integer count = input.getIntegerByField("count");

        Integer sum = counts.get(wordStr);

        if (sum == null) {
            counts.put(wordStr, count);
        } else {
            counts.put(wordStr, sum + count);
        }

        counts.forEach((k, v) -> {
            System.out.println(k + "_________" + v);
        });
        System.out.println("========================================");
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {

    }
}

4、運行

右鍵運行WordCountTopology

二、Storm整體架構

此外，storm/log4j2/cluster.xml文件中可以配置storm的日志級別矩陣信息等。

操作系統(tǒng)的配置，其中有兩項需要配置（通過ulimit?-a查看）：

1、open?files：當前用戶可以打開的文件描述符數(shù)。

2、max?user processes：當前用戶可以運行的進程數(shù)，此參數(shù)太小將引起storm的一個錯誤：

java.lang.OutOfMemoryError: unable to create new native thread

部署注意事項：

1. 在storm根目錄下有一個lib目錄，存放storm本身依賴的jar包，此處的所有jar會被storm?worker進行啟動時加載，個人編寫的jar包不能放在此處，以免包更新帶來不便
2. 向storm集群提交拓撲時，建議將該拓撲所有依賴的jar包和業(yè)務源代碼打到一個jar包中（fat?jar），如此則業(yè)務需要的jar包都和拓撲在同一個jar包中，否則當拓撲依賴的jar包更新時需要將該更新包放到所有的storm節(jié)點上。如果是在一個集群中，fat?jar可以保證不同業(yè)務的jar包是獨立的，不會混淆。

nimbus

1. 接收jar包：提交應用拓撲
2. 任務分配：將拓撲的任務分配給worker
3. 資源調度：監(jiān)控各個supervisor節(jié)點的狀態(tài)進行負載均衡等。
4. Nimbus不需要像supervisor節(jié)點那么高的配置，storm ui也不需要高配置，可以和nimbus節(jié)點運行在同一臺服務器節(jié)點上。

supervisor

1. 監(jiān)聽nimbus的任務分配，啟動分配到的worker來對相應的任務進行處理。
2. 監(jiān)控本地的worker進程，如果發(fā)現(xiàn)狀態(tài)不正常會殺死worker并重啟，超過一定次數(shù)后將分配給該錯誤狀態(tài)的worker的任務交還給nimbus進行再次分配。
3. 刪除本地不再運行的任務

worker

完成拓撲中定義的業(yè)務邏輯，即執(zhí)行拓撲的進程。

一個worker的基本執(zhí)行步驟：

1. 根據(jù)zookeeper中拓撲的組件分配變化，創(chuàng)建或移除worker到worker的鏈接
2. 創(chuàng)建executor（執(zhí)行器）的輸入隊列receive-queue-map和輸出隊列transfer-queue
3. 創(chuàng)建worker的接收線程receive-thread和發(fā)送線程transfer-thread
4. 根據(jù)組件分配關系創(chuàng)建executor
   1. executor即worker JVM進程中的一個java線程，一般默認每個executor負責執(zhí)行一個task任務
5. 在executor中執(zhí)行具體的任務（spout或者bolt）來執(zhí)行具體的業(yè)務邏輯。
   1. 檢查需要運行的task信息
   2. 獲取相應的task信息，即spout/bolt信息

每個任務對應一個線程或多個任務對應一個線程

線程稱為executor

executor在worker中運行

worker是一個JVM進程

在supervisor中運行

worker中的數(shù)據(jù)流：

worker中線程間通信使用的是Disruptor，進程間通信可能是netty也可以是zmq。默認使用netty。

數(shù)據(jù)流：

1. 每個worker綁定一個socket端口作為數(shù)據(jù)的輸入，此端口作為socket的服務器端一直監(jiān)聽運行。
2. 根據(jù)拓撲的關系，確定需要向外通信的任務所在的worker地址，并同該worker也創(chuàng)建好socket連接，此時該worker是作為socket的客戶端。
3. receive thread負責將每個executor所需要的數(shù)據(jù)放入對應的receive-queue-map中，然后由executor來獲取自己所需要的數(shù)據(jù)，這個過程通過disruptor進行通信。
4. executor執(zhí)行完操作需要對外發(fā)送數(shù)據(jù)時，首先kryo將數(shù)據(jù)序列化，然后通過disruptor將數(shù)據(jù)放入對外的transfer-queue中。
5. transfer?thread完成數(shù)據(jù)的發(fā)送工作。
6. 如果executor需要對外發(fā)送的數(shù)據(jù)接收方和executor在同一個worker節(jié)點，則不需要執(zhí)行序列化操作，調用disruptor的publish方法直接放到接收方的executor對應的隊列中即可。

與MapReduce架構的對比：

提交作業(yè)過程

1. 客戶端提交拓撲代碼到nimbus的nimbus/inbox目錄下。
2. nimbus對topology進行校驗、處理
3. nimbus針對該拓撲建立本地目錄：nimbus/stormdist/topology-id

該目錄下有三個文件：文章來源地址http://www.zghlxwxcb.cn/news/detail-715488.html

1. 1. stormjar.jar 從nimbus/inbox移動來的topology的jar包
   2. stormcode.ser 對topology對象的序列化
   3. stormconf.ser topology的運行配置信息
2. nimbus的調度器根據(jù)拓撲的配置計算task，并把task分配到不同的worker上，調度的結果寫入zookeeper的/task節(jié)點下。
3. zookeeper上建立assignments節(jié)點，存儲task和supervisor中worker的對應關系。
4. zookeeper上創(chuàng)建workerbeats節(jié)點監(jiān)控worker的心跳。
5. supervisor去zookeeper上獲取分配的task信息，啟動一個或多個worker來執(zhí)行。
6. 每個worker上運行多個task，task由executor來執(zhí)行。
7. worker根據(jù)拓撲信息初始化建立task之間的連接
8. 相同worker內的task通過DisruptorQueue通信，不同worker間默認采用netty通信

到了這里，關于大數(shù)據(jù)-Storm流式框架（二）--wordcount案例的文章就介紹完了。如果您還想了解更多內容，請在右上角搜索TOY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關文章，希望大家以后多多支持TOY模板網(wǎng)！