storm中如何自定義數(shù)據(jù)分組

今天就跟大家聊聊有關(guān)storm中如何自定義數(shù)據(jù)分組，可能很多人都不太了解，為了讓大家更加了解，小編給大家總結(jié)了以下內(nèi)容，希望大家根據(jù)這篇文章可以有所收獲。

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

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

storm自帶數(shù)據(jù)流組

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

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

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

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

域數(shù)據(jù)流組允許你基于元組的一個或多個域控制如何把元組發(fā)送給bolts。它保證擁有相同域組合的值集發(fā)送給同一個bolt。回到單詞計數(shù)器的例子，如果你用word域為數(shù)據(jù)流分組，word-normalizer bolt將只會把相同單詞的元組發(fā)送給同一個word-counterbolt實例。

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

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

全部數(shù)據(jù)流組，為每個接收數(shù)據(jù)的實例復(fù)制一份元組副本。這種分組方式用于向bolts發(fā)送信號。比如，你要刷新緩存，你可以向所有的bolts發(fā)送一個刷新緩存信號。在單詞計數(shù)器的例子里，你可以使用一個全部數(shù)據(jù)流組，添加清除計數(shù)器緩存的功能 

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

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

這是一個特殊的數(shù)據(jù)流組，數(shù)據(jù)源可以用它決定哪個組件接收元組

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

。與前面的例子類似，數(shù)據(jù)源將根據(jù)單詞首字母決定由哪個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));
            }
        }
        //對元組做出應(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方法中計算任務(wù)數(shù)

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

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

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

不分組

這個數(shù)據(jù)流組相當于隨機數(shù)據(jù)流組。也就是說，使用這個數(shù)據(jù)流組時，并不關(guān)心數(shù)據(jù)流是如何分組的。

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

storm自定義數(shù)據(jù)流組和hadoop Partitioner分組很相似，storm自定義分組要實現(xiàn)CustomStreamGrouping接口，接口源碼如下：

targetTasks就是Storm運行時告訴你，當前有幾個目標Task可以選擇，每一個都給編上了數(shù)字編號。而 chooseTasks(int taskId, List values); 就是讓你選擇，你的這條數(shù)據(jù)values，是要哪幾個目標Task處理？

這是我寫的一個自定義分組，總是把數(shù)據(jù)分到第一個Task：

從上面的代碼可以看出，該自定義分組會把數(shù)據(jù)歸并到第一個TaskArrays.asList(tasks.get(0));，也就是數(shù)據(jù)到達后總是被派發(fā)到第一組。和Hadoop不同的是，Storm允許一條數(shù)據(jù)被多個Task處理，因此返回值是List .就是讓你來在提供的 'List targetTasks' Task中選擇任意的幾個（必須至少是一個）Task來處理數(shù)據(jù)。

第二個自定義分組，wordcount中使首字母相同的單詞交給同一個bolt處理：

public class ModuleGrouping implements CustormStreamGrouping{
        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();
        }
    }

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

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

看完上述內(nèi)容，你們對storm中如何自定義數(shù)據(jù)分組有進一步的了解嗎？如果還想了解更多知識或者相關(guān)內(nèi)容，請關(guān)注億速云行業(yè)資訊頻道，感謝大家的支持。