Chinaunix首页 | 论坛 | 博客
  • 博客访问: 3228401
  • 博文数量: 346
  • 博客积分: 10189
  • 博客等级: 上将
  • 技术积分: 3125
  • 用 户 组: 普通用户
  • 注册时间: 2008-08-05 19:46
文章分类

全部博文(346)

文章存档

2013年(35)

2011年(35)

2010年(76)

2009年(48)

2008年(152)

分类: 大数据

2013-09-13 09:12:26

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

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,核心代码如下:

复制代码
    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() { 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类似,只是将各个RankObjectsBolt合并后计算得到最终全局的TOP N结果,核心代码如下:

复制代码
    public void execute(Tuple tuple, BasicOutputCollector collector) {
        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() { 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"));
    }
复制代码

关于上述例子的几点说明:

(1) 为什么要有RankObjectsMergeObjects两级的Bolt来计算呢?

其实,计算TOP N的一个最简单的思路是直接使用一个Bolt(通过类似于RankObjects的类实现)来做全局的求TOP N操作。

但是,这种方式的明显缺点在于受限于单台机器的处理能力。

(2) 如何保证计算结果的正确性?

首先通过field grouping将同一个word的计算放到同一个Bolt上处理;最后有一个全局的global grouping汇总得到TOP N

这样可以做到最大可能并行性,同时也能保证计算结果的正确。

(3) 如果当前计算资源无法满足计算TOP N,该怎么办?

这个问题本质上就是系统的可扩展性问题,基本的解决方法就是尽可能做到在多个机器上的并行计算过程,针对上面的Topology结构:

a) 可以通过增加每一级处理单元Bolt的数量,减少每个Bolt处理的数据规模;

b) 可以通过增加一级或多级Bolt处理单元,减少最终汇总处理的数据规模。

本文参考代码见:

阅读(2664) | 评论(0) | 转发(0) |
给主人留下些什么吧!~~