碼上快樂

相關內容    簡體    繁體

storm入門（二）：關於storm中某一段時間內topN的計算入門

本文轉載自   查看原文   2015-05-25 16:25   2801    storm

剛剛接觸storm 對於滑動窗口的topN復雜模型有一些不理解，通過閱讀其他的博客發現有兩篇關於topN的非滑動窗口的介紹。然后轉載過來。

下面是第一種：

Storm的另一種常見模式是對流式數據進行所謂“streaming top N”的計算，它的特點是持續的在內存中按照某個統計指標（如出現次數）計算TOP N，然后每隔一定時間間隔輸出實時計算后的TOP N結果。

流式數據的TOP N計算的應用場景很多，例如計算twitter上最近一段時間內的熱門話題、熱門點擊圖片等等。

下面結合Storm-Starter中的例子，介紹一種可以很容易進行擴展的實現方法：首先，在多台機器上並行的運行多個Bolt，每個Bolt負責一部分數據的TOP N計算，然后再有一個全局的Bolt來合並這些機器上計算出來的TOP N結果，合並后得到最終全局的TOP N結果。

該部分示例代碼的入口是RollingTopWords類，用於計算文檔中出現次數最多的N個單詞。首先看一下這個Topology結構：

 

Topology構建的代碼如下：

  
        TopologyBuilder builder = new TopologyBuilder();
        builder.setSpout("word", new TestWordSpout(), 5);
        builder.setBolt("count", new RollingCountObjects(60, 10), 4)
                 .fieldsGrouping("word", new Fields("word"));
        builder.setBolt("rank", new RankObjects(TOP_N), 4)
                 .fieldsGrouping("count", new Fields("obj"));
        builder.setBolt("merge", new MergeObjects(TOP_N))
                 .globalGrouping("rank");
 

 

（1）首先，TestWordSpout()是Topology的數據源Spout，持續隨機生成單詞發出去，產生數據流“word”，輸出Fields是“word”，核心代碼如下：

  
    public void nextTuple() {
        Utils.sleep(100);
        final String[] words = new String[] {"nathan", "mike", "jackson", "golda", "bertels"};
        final Random rand = new Random();
        final String word = words[rand.nextInt(words.length)];
        _collector.emit(new Values(word));
　　}
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("word"));
　　}
 

 

（2）接下來，“word”流入RollingCountObjects這個Bolt中進行word count計算，為了保證同一個word的數據被發送到同一個Bolt中進行處理，按照“word”字段進行field grouping；在RollingCountObjects中會計算各個word的出現次數，然后產生“count”流，輸出“obj”和“count”兩個Field，其中對於synchronized的線程鎖我們也可以換成安全的容器，比如ConcurrentHashMap等組件。核心代碼如下：

  
    public void execute(Tuple tuple) {

        Object obj = tuple.getValue(0);
        int bucket = currentBucket(_numBuckets);
        synchronized(_objectCounts) {
            long[] curr = _objectCounts.get(obj);
            if(curr==null) {
                curr = new long[_numBuckets];
                _objectCounts.put(obj, curr);
            }
            curr[bucket]++;
            _collector.emit(new Values(obj, totalObjects(obj)));
            _collector.ack(tuple);
        }
    }
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("obj", "count"));
    }
 

（3）然后，RankObjects這個Bolt按照“count”流的“obj”字段進行field grouping；在Bolt內維護TOP N個有序的單詞，如果超過TOP N個單詞，則將排在最后的單詞踢掉，同時每個一定時間（2秒）產生“rank”流，輸出“list”字段，輸出TOP N計算結果到下一級數據流“merge”流，核心代碼如下：

  
    public void execute(Tuple tuple, BasicOutputCollector collector) {
        Object tag = tuple.getValue(0);
        Integer existingIndex = _find(tag);
        if (null != existingIndex) {
            _rankings.set(existingIndex, tuple.getValues());
        } else {
            _rankings.add(tuple.getValues());
        }
        Collections.sort(_rankings, new Comparator<List>() {
            public int compare(List o1, List o2) {
                return _compare(o1, o2);
            }
        });
        if (_rankings.size() > _count) {
            _rankings.remove(_count);
        }
        long currentTime = System.currentTimeMillis();
        if(_lastTime==null || currentTime >= _lastTime + 2000) {
            collector.emit(new Values(new ArrayList(_rankings)));
            _lastTime = currentTime;
        }
    }

    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("list"));
    }
 

（4）最后，MergeObjects這個Bolt按照“rank”流的進行全局的grouping，即所有上一級Bolt產生的“rank”流都流到這個“merge”流進行；MergeObjects的計算邏輯和RankObjects類似，只是將各個RankObjects的Bolt合並后計算得到最終全局的TOP N結果，核心代碼如下：

  
    public void execute(Tuple tuple, BasicOutputCollector collector) {
        List<List> merging = (List) tuple.getValue(0);
        for(List pair : merging) {
            Integer existingIndex = _find(pair.get(0));
            if (null != existingIndex) {
                _rankings.set(existingIndex, pair);
            } else {
                _rankings.add(pair);
            }

            Collections.sort(_rankings, new Comparator<List>() {
                public int compare(List o1, List o2) {
                    return _compare(o1, o2);
                }
            });

            if (_rankings.size() > _count) {
                _rankings.subList(_count, _rankings.size()).clear();
            }
        }

        long currentTime = System.currentTimeMillis();
        if(_lastTime==null || currentTime >= _lastTime + 2000) {
            collector.emit(new Values(new ArrayList(_rankings)));
            LOG.info("Rankings: " + _rankings);
            _lastTime = currentTime;
        }
    }

    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("list"));
    }
 

另外，還有一種很聰明的方法，只在execute中插入數據而不emit，而在prepare中進行emit，創建線程根據時間進行監聽。

  1. package test.storm.topology;
  3. import test.storm.bolt.WordCounter;
  4. import test.storm.bolt.WordWriter;
  5. import test.storm.spout.WordReader;
  6. import backtype.storm.Config;
  7. import backtype.storm.StormSubmitter;
  8. import backtype.storm.generated.AlreadyAliveException;
  9. import backtype.storm.generated.InvalidTopologyException;
  10. import backtype.storm.topology.TopologyBuilder;
  11. import backtype.storm.tuple.Fields;
  13. public class WordTopN {
  14.     public static void main(String[] args) throws AlreadyAliveException, InvalidTopologyException {
  15.         if (args == null || args.length < 1) {　　
  16.             System.err.println("Usage: N");
  17.             System.err.println("such as : 10");
  18.             System.exit(-1);
  19.         }
  21.         TopologyBuilder builder = new TopologyBuilder();
  22.         builder.setSpout("wordreader", new WordReader(), 2);
  23.         builder.setBolt("wordcounter", new WordCounter(), 2).fieldsGrouping("wordreader", new Fields("word"));
  24.         builder.setBolt("wordwriter", new WordWriter()).globalGrouping("wordcounter");
  26.         Config conf = new Config();
  27.         conf.put("N", args[0]);
  29.         conf.setDebug(false);
  30.         StormSubmitter.submitTopology("topN", conf, builder.createTopology());
  32.     }
  33. }

這里需要注意的幾點是，第一個bolt的分組策略是fieldsGrouping，按照字段分組，這一點很重要，它能保證相同的word被分發到同一個bolt上，
像做wordcount、TopN之類的應用就要使用這種分組策略。
最后一個bolt的分組策略是globalGrouping，全局分組，tuple會被分配到一個bolt用來匯總。
為了提高並行度，spout和第一個bolt均設置並行度為2（我這里測試機器性能不是很高）。

點擊(此處)折疊或打開

  1. package test.storm.spout;
  3. import java.util.Map;
  4. import java.util.Random;
  5. import java.util.concurrent.atomic.AtomicInteger;
  7. import backtype.storm.spout.SpoutOutputCollector;
  8. import backtype.storm.task.TopologyContext;
  9. import backtype.storm.topology.OutputFieldsDeclarer;
  10. import backtype.storm.topology.base.BaseRichSpout;
  11. import backtype.storm.tuple.Fields;
  12. import backtype.storm.tuple.Values;
  14. public class WordReader extends BaseRichSpout {
  15.     private static final long serialVersionUID = 2197521792014017918L;
  16.     private SpoutOutputCollector collector;
  17.     private static AtomicInteger i = new AtomicInteger();
  18.     private static String[] words = new String[] { \"a\", \"b\", \"c\", \"d\", \"e\", \"f\", \"g\", \"h\", \"i\", \"j\", \"k\", \"l\", \"m\",
  19.             \"n\", \"o\", \"p\", \"q\", \"r\", \"s\", \"t\", \"u\", \"v\", \"w\", \"x\", \"y\", \"z\" };
  21.     @Override
  22.     public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
  23.         this.collector = collector;
  24.     }
  26.     @Override
  27.     public void nextTuple() {
  28.         if (i.intValue() < 100) {
  29.             Random rand = new Random();
  30.             String word = words[rand.nextInt(words.length)];
  31.             collector.emit(new Values(word));
  32.             i.incrementAndGet();
  33.         }
  34.     }
  36.     @Override
  37.     public void declareOutputFields(OutputFieldsDeclarer declarer) {
  38.         declarer.declare(new Fields("word"));
  39.     }
  40. }

spout的作用是隨機發送word，發送100次，由於並行度是2，將產生2個spout實例，所以這里的計數器使用了static的AtomicInteger來保證線程安全。


點擊(此處)折疊或打開

  1. package test.storm.bolt;
  3. import java.util.ArrayList;
  4. import java.util.Collections;
  5. import java.util.Comparator;
  6. import java.util.HashMap;
  7. import java.util.List;
  8. import java.util.Map;
  9. import java.util.Map.Entry;
  10. import java.util.concurrent.ConcurrentHashMap;
  12. import backtype.storm.task.OutputCollector;
  13. import backtype.storm.task.TopologyContext;
  14. import backtype.storm.topology.IRichBolt;
  15. import backtype.storm.topology.OutputFieldsDeclarer;
  16. import backtype.storm.tuple.Fields;
  17. import backtype.storm.tuple.Tuple;
  18. import backtype.storm.tuple.Values;
  20. public class WordCounter implements IRichBolt {
  21.     private static final long serialVersionUID = 5683648523524179434L;
  22.     private static Map<String, Integer> counters = new ConcurrentHashMap<String, Integer>();
  23.     private volatile boolean edit = true;
  25.     @Override
  26.     public void prepare(final Map stormConf, TopologyContext context, final OutputCollector collector) {
  27.         new Thread(new Runnable() {
  28.             @Override
  29.             public void run() {
  30.                 while (true) {
  31.                     //5秒后counter不再變化，可以認為spout已經發送完畢
  32.                     if (!edit) {
  33.                         if (counters.size() > 0) {
  34.                             List<Map.Entry<String, Integer>> list = new ArrayList<Map.Entry<String, Integer>>();
  35.                             list.addAll(counters.entrySet());
  36.                             Collections.sort(list, new ValueComparator());
  38.                             //向下一個bolt發送前N個word
  39.                             for (int i = 0; i < list.size(); i++) {
  40.                                 if (i < Integer.parseInt(stormConf.get("N").toString())) {
  41.                                     collector.emit(new Values(list.get(i).getKey() + ":" + list.get(i).getValue()));
  42.                                 }
  43.                             }
  44.                         }
  46.                         //發送之后，清空counters，以防spout再次發送word過來
  47.                         counters.clear();
  48.                     }
  50.                     edit = false;
  51.                     try {
  52.                         Thread.sleep(5000);
  53.                     } catch (InterruptedException e) {
  54.                         e.printStackTrace();
  55.                     }
  56.                 }
  57.             }
  58.         }).start();
  59.     }
  61.     @Override
  62.     public void execute(Tuple tuple) {
  63.         String str = tuple.getString(0);
  64.         if (counters.containsKey(str)) {
  65.             Integer c = counters.get(str) + 1;
  66.             counters.put(str, c);
  67.         } else {
  68.             counters.put(str, 1);
  69.         }
  71.         edit = true;
  72.     }
  74.     private static class ValueComparator implements Comparator<Map.Entry<String, Integer>> {
  75.         @Override
  76.         public int compare(Entry<String, Integer> entry1, Entry<String, Integer> entry2) {
  77.             return entry2.getValue() - entry1.getValue();
  78.         }
  79.     }
  81.     @Override
  82.     public void declareOutputFields(OutputFieldsDeclarer declarer) {
  83.         declarer.declare(new Fields("word_count"));
  84.     }
  86.     @Override
  87.     public void cleanup() {
  88.     }
  90.     @Override
  91.     public Map<String, Object> getComponentConfiguration() {
  92.         return null;
  93.     }
  94. }

在WordCounter里面有個線程安全的容器ConcurrentHashMap，來存儲word以及對應的次數。在prepare方法里啟動一個線程，長期監聽edit的狀態，監聽間隔是5秒，
當edit為false，即execute方法不再執行、容器不再變化，可以認為spout已經發送完畢了，可以開始排序取TopN了。這里使用了一個volatile edit（回憶一下volatile的使用場景：
對變量的修改不依賴變量當前的值，這里設置true or false，顯然不相互依賴）。


點擊(此處)折疊或打開

  1. package test.storm.bolt;
  3. import java.io.FileWriter;
  4. import java.io.IOException;
  5. import java.util.Map;
  7. import backtype.storm.task.TopologyContext;
  8. import backtype.storm.topology.BasicOutputCollector;
  9. import backtype.storm.topology.OutputFieldsDeclarer;
  10. import backtype.storm.topology.base.BaseBasicBolt;
  11. import backtype.storm.tuple.Tuple;
  13. public class WordWriter extends BaseBasicBolt {
  14.     private static final long serialVersionUID = -6586283337287975719L;
  15.     private FileWriter writer = null;
  17.     public WordWriter() {
  18.     }
  20.     @Override
  21.     public void prepare(Map stormConf, TopologyContext context) {
  22.         try {
  23.             writer = new FileWriter("/data/tianzhen/output/" + this);
  24.         } catch (IOException e) {
  25.             e.printStackTrace();
  26.         }
  27.     }
  29.     @Override
  30.     public void execute(Tuple input, BasicOutputCollector collector) {
  31.         String s = input.getString(0);
  32.         try {
  33.             writer.write(s);
  34.             writer.write("\n");
  35.             writer.flush();
  36.         } catch (IOException e) {
  37.             e.printStackTrace();
  38.         } finally {
  39.             //writer不能close，因為execute需要一直運行
  40.         }
  41.     }
  43.     @Override
  44.     public void declareOutputFields(OutputFieldsDeclarer declarer) {
  46.     }
  47. }

最后一個bolt做全局的匯總，這里我偷了懶，直接將結果寫到文件了，省略截取TopN的過程，因為我這里就一個supervisor節點，所以結果是正確的。

引用連接：http://blog.itpub.net/28912557/viewspace-1579860/

　　　　　http://www.cnblogs.com/panfeng412/archive/2012/06/16/storm-common-patterns-of-streaming-top-n.html


免責聲明！

本站轉載的文章為個人學習借鑒使用，本站對版權不負任何法律責任。如果侵犯了您的隱私權益，請聯系本站郵箱yoyou2525@163.com刪除。



猜您在找 postgre查詢一段時間內的數據 js獲取一段時間內的間隔日期 截取一段時間內的log日志 Pgsql之查詢一段時間內的所有日期 用sql獲取一段時間內的數據 如何使用 SVN 找到一段時間內提交的代碼文件 python 缺失值處理 - 獲取一段時間內所有日期的列表 Pgsql之查詢一段時間內的所有年月yyyy-mm [原創]JS端計算一段時間內工作日的天數，排除周末和法定節假日，同時考慮到調休日 PHP獲取一段時間內的每個周幾, 每月幾號, 遇到特殊日子就往后延
 
粵ICP備18138465號   © 2018-2025 CODEPRJ.COM