首页 \ 教程 \ storm

知识点

Storm

Twitter Storm常见模式

Storm常见模式——批处理

hibernate 1+N 问题

Storm常见问题(笔记)

如果再不改变你就变成了N年后的SB

Storm常见模式——BasicBolt

物理专业英语词汇(H-N)

Storm常见模式——流聚合

Storm常见模式——TimeCacheMap

Storm常见模式——分布式RPC

I18n的一个问题

初试Twitter Storm之常见问题

hibernate一级缓存及N+1问题

solr 优化索引成 2段或N段

I18N 国际化简介

Storm常见模式——求TOP N

2019-03-02 23:44|来源: 网路

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<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"));
    }

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

（1）为什么要有RankObjects和MergeObjects两级的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处理单元，减少最终汇总处理的数据规模。

本文参考代码见：https://github.com/nathanmarz/storm-starter

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

知识点

相关文章

最近更新

Storm常见模式——求TOP N

相关问答

编程求n！，n由键盘输入[2022-04-15]

用类和对象实现求n的阶乘[2023-12-18]

什么是[-n“$ VARIABLE”] ||(what does [ -n “$VARIABLE” ] || exit 0 mean)[2021-11-27]

Python：如何使用re {n}(Python: how to use re{n})[2023-06-08]

n＆〜（n - 1）这个函数做什么？(n & ~(n - 1) What does this function do?)[2022-10-11]

求解递归W（n）= W（n / 2）+ n log n？(Solving the recurrence W(n) = W(n / 2) + n log n?)[2022-05-12]

给定2 ^ n，使用对数求n(Given 2^n, find n using logarithm)[2022-03-01]

java ArrayList中最常见的n - 单词(Most frequent n - words in java ArrayList)[2022-05-29]

apache storm topology id从n跳到n + 2(apache storm topology id skips from n to n+2)[2022-03-29]

如何强制emacs使用\ n而不是\ r \ n(How to force emacs to use \n instead of \r\n)[2023-04-23]