Storm的Topology怎么配置

這篇文章主要介紹“Storm的Topology怎么配置”，在日常操作中，相信很多人在Storm的Topology怎么配置問(wèn)題上存在疑惑，小編查閱了各式資料，整理出簡(jiǎn)單好用的操作方法，希望對(duì)大家解答”Storm的Topology怎么配置”的疑惑有所幫助！接下來(lái)，請(qǐng)跟著小編一起來(lái)學(xué)習(xí)吧！

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

設(shè)計(jì)一個(gè)拓?fù)鋾r(shí)，你要做的最重要的事情之一就是定義如何在各組件之間交換數(shù)據(jù)（數(shù)據(jù)流是如何被bolts消費(fèi)的）。一個(gè)數(shù)據(jù)流組指定了每個(gè)bolt會(huì)消費(fèi)哪些數(shù)據(jù)流，以及如何消費(fèi)它們。

NOTE：一個(gè)節(jié)點(diǎn)能夠發(fā)布一個(gè)以上的數(shù)據(jù)流，一個(gè)數(shù)據(jù)流組允許我們選擇接收哪個(gè)。

數(shù)據(jù)流組在定義拓?fù)鋾r(shí)設(shè)置，就像我們?cè)诘诙驴吹降模?/p>

···
    builder.setBolt("word-normalizer", new WordNormalizer())
           .shuffleGrouping("word-reader");
···

在前面的代碼塊里，一個(gè)bolt由TopologyBuilder對(duì)象設(shè)定， 然后使用隨機(jī)數(shù)據(jù)流組指定數(shù)據(jù)源。數(shù)據(jù)流組通常將數(shù)據(jù)源組件的ID作為參數(shù)，取決于數(shù)據(jù)流組的類型不同還有其它可選參數(shù)。

NOTE：每個(gè)InputDeclarer可以有一個(gè)以上的數(shù)據(jù)源，而且每個(gè)數(shù)據(jù)源可以分到不同的組。

隨機(jī)數(shù)據(jù)流組（隨機(jī)分組）

隨機(jī)流組是最常用的數(shù)據(jù)流組。它只有一個(gè)參數(shù)（數(shù)據(jù)源組件），并且數(shù)據(jù)源會(huì)向隨機(jī)選擇的bolt發(fā)送元組，保證每個(gè)消費(fèi)者收到近似數(shù)量的元組。

隨機(jī)數(shù)據(jù)流組用于數(shù)學(xué)計(jì)算這樣的原子操作。然而，如果操作不能被隨機(jī)分配，就像第二章為單詞計(jì)數(shù)的例子，你就要考慮其它分組方式了。

域數(shù)據(jù)流組（字段分組）

域數(shù)據(jù)流組允許你基于元組的一個(gè)或多個(gè)域控制如何把元組發(fā)送給bolts。它保證擁有相同域組合的值集發(fā)送給同一個(gè)bolt?；氐絾卧~計(jì)數(shù)器的例子，如果你用word域?yàn)閿?shù)據(jù)流分組，word-normalizer bolt將只會(huì)把相同單詞的元組發(fā)送給同一個(gè)word-counterbolt實(shí)例。

···
    builder.setBolt("word-counter", new WordCounter(),2)
           .fieldsGrouping("word-normalizer", new Fields("word"));
···

NOTE: 在域數(shù)據(jù)流組中的所有域集合必須存在于數(shù)據(jù)源的域聲明中。

全部數(shù)據(jù)流組（全部分組）

全部數(shù)據(jù)流組，為每個(gè)接收數(shù)據(jù)的實(shí)例復(fù)制一份元組副本。這種分組方式用于向bolts發(fā)送信號(hào)。比如，你要刷新緩存，你可以向所有的bolts發(fā)送一個(gè)刷新緩存信號(hào)。在單詞計(jì)數(shù)器的例子里，你可以使用一個(gè)全部數(shù)據(jù)流組，添加清除計(jì)數(shù)器緩存的功能（見(jiàn)拓?fù)涫纠?/p>

    public void execute(Tuple input) {
        String str = null;
        try{
            if(input.getSourceStreamId().equals("signals")){
                str = input.getStringByField("action");
                if("refreshCache".equals(str))
                    counters.clear();
            }
        }catch (IllegalArgumentException e){
            //什么也不做
        }
        ···
    }

我們添加了一個(gè)if分支，用來(lái)檢查源數(shù)據(jù)流。Storm允許我們聲明具名數(shù)據(jù)流（如果你不把元組發(fā)送到一個(gè)具名數(shù)據(jù)流，默認(rèn)發(fā)送到名為”default“的數(shù)據(jù)流）。這是一個(gè)識(shí)別元組的極好的方式，就像這個(gè)例子中，我們想識(shí)別signals一樣。 在拓?fù)涠x中，你要向word-counter bolt添加第二個(gè)數(shù)據(jù)流，用來(lái)接收從signals-spout數(shù)據(jù)流發(fā)送到所有bolt實(shí)例的每一個(gè)元組。

   builder.setBolt("word-counter", new WordCounter(),2)
           .fieldsGroupint("word-normalizer",new Fields("word"))
           .allGrouping("signals-spout","signals");

signals-spout的實(shí)現(xiàn)請(qǐng)參考git倉(cāng)庫(kù)。

自定義數(shù)據(jù)流組（自定義分組）

你可以通過(guò)實(shí)現(xiàn)backtype.storm.grouping.CustormStreamGrouping接口創(chuàng)建自定義數(shù)據(jù)流組，讓你自己決定哪些bolt接收哪些元組。

讓我們修改單詞計(jì)數(shù)器示例，使首字母相同的單詞由同一個(gè)bolt接收。

public class ModuleGrouping mplents CustormStreamGrouping, Serializable{
        int numTasks = 0;
        @Override
        public List<Integer> chooseTasks(List<Object> values) {
            List<Integer> boltIds = new ArrayList<Integer>();
            if(values.size()>0){
                String str = values.get(0).toString();
                if(str.isEmpty()){
                    boltIds.add(0);
                }else{
                    boltIds.add(str.charAt(0) % numTasks);
                }
            }
            return boltIds;
        }
        @Override
        public void prepare(TopologyContext context, Fields outFields, List<Integer> targetTasks) {
            numTasks = targetTasks.size();
        }
    }

這是一個(gè)CustomStreamGrouping的簡(jiǎn)單實(shí)現(xiàn)，在這里我們采用單詞首字母字符的整數(shù)值與任務(wù)數(shù)的余數(shù)，決定接收元組的bolt。

按下述方式word-normalizer修改即可使用這個(gè)自定義數(shù)據(jù)流組。

 builder.setBolt("word-normalizer", new WordNormalizer())
           .customGrouping("word-reader", new ModuleGrouping());

直接數(shù)據(jù)流組（直接分組）

這是一個(gè)特殊的數(shù)據(jù)流組，數(shù)據(jù)源可以用它決定哪個(gè)組件接收元組。與前面的例子類似，數(shù)據(jù)源將根據(jù)單詞首字母決定由哪個(gè)bolt接收元組。要使用直接數(shù)據(jù)流組，在WordNormalizer bolt中，使用emitDirect方法代替emit。

public void execute(Tuple input) {
        ...
        for(String word : words){
            if(!word.isEmpty()){
                ...
                collector.emitDirect(getWordCountIndex(word),new Values(word));
            }
        }
        //對(duì)元組做出應(yīng)答
        collector.ack(input);
    }
    public Integer getWordCountIndex(String word) {
        word = word.trim().toUpperCase();
        if(word.isEmpty()){
            return 0;
        }else{
            return word.charAt(0) % numCounterTasks;
        }
    }

在prepare方法中計(jì)算任務(wù)數(shù)

public void prepare(Map stormConf, TopologyContext context, 
                OutputCollector collector) {
        this.collector = collector;
        this.numCounterTasks = context.getComponentTasks("word-counter");
    }

在拓?fù)涠x中指定數(shù)據(jù)流將被直接分組：

builder.setBolt("word-counter", new WordCounter(),2)
           .directGrouping("word-normalizer");

全局?jǐn)?shù)據(jù)流組（全局分組）

全局?jǐn)?shù)據(jù)流組把所有數(shù)據(jù)源創(chuàng)建的元組發(fā)送給單一目標(biāo)實(shí)例（即擁有最低ID的任務(wù)）。

不分組（無(wú)分組）

寫(xiě)作本書(shū)時(shí)（Stom0.7.1版），這個(gè)數(shù)據(jù)流組相當(dāng)于隨機(jī)數(shù)據(jù)流組。也就是說(shuō)，使用這個(gè)數(shù)據(jù)流組時(shí)，并不關(guān)心數(shù)據(jù)流是如何分組的。

LocalCluster VS StormSubmitter

到目前為止，你已經(jīng)用一個(gè)叫做LocalCluster的工具在你的本地機(jī)器上運(yùn)行了一個(gè)拓?fù)?。Storm的基礎(chǔ)工具，使你能夠在自己的計(jì)算機(jī)上方便的運(yùn)行和調(diào)試不同的拓?fù)?。但是你怎么把自己的拓?fù)涮峤唤o運(yùn)行中的Storm集群呢？Storm有一個(gè)有趣的功能，在一個(gè)真實(shí)的集群上運(yùn)行自己的拓?fù)涫呛苋菀椎氖虑?。要?shí)現(xiàn)這一點(diǎn)，你需要把LocalCluster換成StormSubmitter并實(shí)現(xiàn)submitTopology方法， 它負(fù)責(zé)把拓?fù)浒l(fā)送給集群。

下面是修改后的代碼：

//LocalCluster cluster = new LocalCluster();
    //cluster.submitTopology("Count-Word-Topology-With-Refresh-Cache", conf, 
    //builder.createTopology());
    StormSubmitter.submitTopology("Count-Word-Topology-With_Refresh-Cache", conf,
            builder.createTopology());
    //Thread.sleep(1000);
    //cluster.shutdown();

NOTE: 當(dāng)你使用StormSubmitter時(shí)，你就不能像使用LocalCluster時(shí)一樣通過(guò)代碼控制集群了。

接下來(lái)，把源碼壓縮成一個(gè)jar包，運(yùn)行Storm客戶端命令，把拓?fù)涮峤唤o集群。如果你已經(jīng)使用了Maven， 你只需要在命令行進(jìn)入源碼目錄運(yùn)行：mvn package。

現(xiàn)在你生成了一個(gè)jar包，使用storm jar命令提交拓?fù)洌P(guān)于如何安裝Storm客戶端請(qǐng)參考附錄A）。命令格式：storm jar allmycode.jar org.me.MyTopology arg1 arg2 arg3。

對(duì)于這個(gè)例子，在拓?fù)涔こ棠夸浵旅孢\(yùn)行：

storm jar target/Topologies-0.0.1-SNAPSHOT.jar countword.TopologyMain src/main/resources/words.txt

通過(guò)這些命令，你就把拓?fù)浒l(fā)布集群上了。

如果想停止或殺死它，運(yùn)行：

storm kill Count-Word-Topology-With-Refresh-Cache

NOTE：拓?fù)涿Q必須保證惟一性。

NOTE：如何安裝Storm客戶端，參考附錄A

DRPC拓?fù)?/strong>

有一種特殊的拓?fù)漕愋徒凶龇植际竭h(yuǎn)程過(guò)程調(diào)用（DRPC），它利用Storm的分布式特性執(zhí)行遠(yuǎn)程過(guò)程調(diào)用（RPC）（見(jiàn)下圖）。Storm提供了一些用來(lái)實(shí)現(xiàn)DRPC的工具。第一個(gè)是DRPC服務(wù)器，它就像是客戶端和Storm拓?fù)渲g的連接器，作為拓?fù)涞?em>spout的數(shù)據(jù)源。它接收一個(gè)待執(zhí)行的函數(shù)和函數(shù)參數(shù)，然后對(duì)于函數(shù)操作的每一個(gè)數(shù)據(jù)塊，這個(gè)服務(wù)器都會(huì)通過(guò)拓?fù)浞峙湟粋€(gè)請(qǐng)求ID用來(lái)識(shí)別RPC請(qǐng)求。拓?fù)鋱?zhí)行最后的bolt時(shí)，它必須分配RPC請(qǐng)求ID和結(jié)果，使DRPC服務(wù)器把結(jié)果返回正確的客戶端。

NOTE：單實(shí)例DRPC服務(wù)器能夠執(zhí)行許多函數(shù)。每個(gè)函數(shù)由一個(gè)惟一的名稱標(biāo)識(shí)。

Storm提供的第二個(gè)工具（已在例子中用過(guò)）是LineDRPCTopologyBuilder，一個(gè)輔助構(gòu)建DRPC拓?fù)涞某橄蟾拍?。生成的拓?fù)鋭?chuàng)建DRPCSpouts——它連接到DRPC服務(wù)器并向拓?fù)涞钠渌糠址职l(fā)數(shù)據(jù)——并包裝bolts，使結(jié)果從最后一個(gè)bolt返回。依次執(zhí)行所有添加到LinearDRPCTopologyBuilder對(duì)象的bolts。

作為這種類型的拓?fù)涞囊粋€(gè)例子，我們創(chuàng)建了一個(gè)執(zhí)行加法運(yùn)算的進(jìn)程。雖然這是一個(gè)簡(jiǎn)單的例子，但是這個(gè)概念可以擴(kuò)展到復(fù)雜的分布式計(jì)算。

bolt按下面的方式聲明輸出：

public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("id","result"));
    }

因?yàn)檫@是拓?fù)渲形┮坏?em>bolt，它必須發(fā)布RPC ID和結(jié)果。execute方法負(fù)責(zé)執(zhí)行加法運(yùn)算。

public void execute(Tuple input) {
        String[] numbers = input.getString(1).split("\\+");
        Integer added = 0;
        if(numbers.length<2){
            throw new InvalidParameterException("Should be at least 2 numbers");
        }
        for(String num : numbers){
            added += Integer.parseInt(num);
        }
        collector.emit(new Values(input.getValue(0),added));
    }

包含加法bolt的拓?fù)涠x如下：

public static void main(String[] args) {
        LocalDRPC drpc = new LocalDRPC();
        LinearDRPCTopologyBuilder builder = new LinearDRPCTopologyBuilder("add");
        builder.addBolt(AdderBolt(),2);
        Config conf = new Config();
        conf.setDebug(true);
        LocalCluster cluster = new LocalCluster();
        cluster.submitTopology("drpcder-topology", conf,
            builder.createLocalTopology(drpc));
        String result = drpc.execute("add", "1+-1");
        checkResult(result,0);
        result = drpc.execute("add", "1+1+5+10");
        checkResult(result,17);
        cluster.shutdown();
        drpc.shutdown();
    }

創(chuàng)建一個(gè)LocalDRPC對(duì)象在本地運(yùn)行DRPC服務(wù)器。接下來(lái)，創(chuàng)建一個(gè)拓?fù)錁?gòu)建器（譯者注：LineDRpctopologyBuilder對(duì)象），把bolt添加到拓?fù)?。運(yùn)行DRPC對(duì)象（LocalDRPC對(duì)象）的execute方法測(cè)試拓?fù)洹?/p>

NOTE：使用DRPCClient類連接遠(yuǎn)程DRPC服務(wù)器。DRPC服務(wù)器暴露了Thrift API，因此可以跨語(yǔ)言編程；并且不論是在本地還是在遠(yuǎn)程運(yùn)行DRPC服務(wù)器，它們的API都是相同的。 對(duì)于采用Storm配置的DRPC配置參數(shù)的Storm集群，調(diào)用構(gòu)建器對(duì)象的createRemoteTopology向Storm集群提交一個(gè)拓?fù)?，而不是調(diào)用createLocalTopology。

到此，關(guān)于“Storm的Topology怎么配置”的學(xué)習(xí)就結(jié)束了，希望能夠解決大家的疑惑。理論與實(shí)踐的搭配能更好的幫助大家學(xué)習(xí)，快去試試吧！若想繼續(xù)學(xué)習(xí)更多相關(guān)知識(shí)，請(qǐng)繼續(xù)關(guān)注億速云網(wǎng)站，小編會(huì)繼續(xù)努力為大家?guī)?lái)更多實(shí)用的文章！