storm RollingTopWords 实时top-N计算任务窗口设计

（2）RollingCountBolt负责统计Window length范围内的所有单词计数并每Sliding interval时间发送一次汇总信息到下游。

  (3)  IntermediateRankingsBolt,这是个中间bolt，主要是为了预先计算部分word的top-N排行榜出来，减少最终节点的排序工作。

（4）TotalRankingsBolt 最终top-N排序并输出计算结果。

1     String spoutId = "wordGenerator";
2     String counterId = "counter";
3     String intermediateRankerId = "intermediateRanker";
4     String totalRankerId = "finalRanker";
5     builder.setSpout(spoutId, new TestWordSpout(), 5);
6     builder.setBolt(counterId, new RollingCountBolt(9, 3), 4).fieldsGrouping(spoutId, new Fields("word"));
7     builder.setBolt(intermediateRankerId, new IntermediateRankingsBolt(TOP_N), 4).fieldsGrouping(counterId, new Fields(
8         "obj"));
9     builder.setBolt(totalRankerId, new TotalRankingsBolt(TOP_N)).globalGrouping(intermediateRankerId);

 RollingCountBolt初始化参数正好就是上面提到的windowLengthInSeconds和emitFrequencyInSeconds，new RollingCountBolt(300, 60)表示每分钟输出一下最近五分钟内的数据统计。

counter = new SlidingWindowCounter<Object>(deriveNumWindowChunksFrom(this.windowLengthInSeconds,this.emitFrequencyInSeconds));

RollingCountBolt内部存放了一个SlidingWindowCounter的结构，SlidingWindowCounter内部存储了SlotBasedCounter，SlotBasedCounter才是具体实现了怎样进行

窗口计算，滑动的窗口不停对应到一个环形的slot列表中。SlidingWindowCounter在窗口滑动的时候采取了如下动作：

  public Map<T, Long> getCountsThenAdvanceWindow() {
    Map<T, Long> counts = objCounter.getCounts();
    objCounter.wipeZeros();
    objCounter.wipeSlot(tailSlot);
    advanceHead();
    return counts;
  }

View Code 1. 取出Map<T, Long> counts, 对象和窗口内所有slots求和值的map 

2. 调用wipeZeros, 删除已经不被使用的obj, 释放空间 
3. 最重要的一步, 清除tailSlot, 并advanceHead, 以实现滑动窗口 
advanceHead的实现, 如何在数组实现循环的滑动窗口

  private void advanceHead() {
    headSlot = tailSlot;
    tailSlot = slotAfter(tailSlot);
  }

SlotBasedCounter主要用于按照窗口对应的slot进行incrementCount，getCounts和computeTotalCount，用于数据新增统计，全窗口数据提取和对应元素窗口内全部slot数据求和。

  public void incrementCount(T obj, int slot) {
    long[] counts = objToCounts.get(obj);
    if (counts == null) {
      counts = new long[this.numSlots];
      objToCounts.put(obj, counts);
    }
    counts[slot]++;
  }


  public Map<T, Long> getCounts() {
    Map<T, Long> result = new HashMap<T, Long>();
    for (T obj : objToCounts.keySet()) {
      result.put(obj, computeTotalCount(obj));
    }
    return result;
  }

  private long computeTotalCount(T obj) {
    long[] curr = objToCounts.get(obj);
    long total = 0;
    for (long l : curr) {
      total += l;
    }
    return total;
  }

如上所述， RollingCountBolt在没有窗口接口的情况下通过代码结构巧妙的实现了一个滑动窗口（理论上滚动窗口也一样可以实现），感觉还是很巧妙地。

参考资料：

1 Storm starter - RollingTopWords