Storm的窗口操作

Storm可同时处理窗口内的所有tuple。窗口可以从时间或数量上来划分，由如下两个因素决定：

窗口的长度(windowLength)，可以是时间间隔或Tuple数量；

滑动间隔(sliding Interval)，可以是时间间隔或Tuple数量；

※要确保topology的过期时间大于窗口的大小加上滑动间隔

Sliding Window：滑动窗口

按照固定的时间间隔或者Tuple数量滑动窗口。

如果滑动间隔和窗口大小一样则等同于滚窗，

如果滑动间隔大于窗口大小则会丢失数据，

如果滑动间隔小于窗口大小则会窗口重叠。

Tumbling Window：滚动窗口

元组被单个窗口处理，一个元组只属于一个窗口，不会有窗口重叠。根据经验其实一般用滚动就可以了

构造builder的时候支持以下的配置

（时间和数量的排列组合）：

withWindow(Count windowLength, Count slidingInterval)
滑窗窗口长度：tuple数, 滑动间隔: tuple数

withWindow(Count windowLength)
滑窗窗口长度：tuple数, 滑动间隔: 每个tuple进来都滑

withWindow(Count windowLength, Duration slidingInterval)
滑窗窗口长度：tuple数, 滑动间隔: 时间间隔

withWindow(Duration windowLength, Duration slidingInterval)
滑窗窗口长度：时间间隔, 滑动间隔: 时间间隔

withWindow(Duration windowLength)
滑窗窗口长度：时间间隔, 滑动间隔: 每个tuple进来都滑

withWindow(Duration windowLength, Count slidingInterval)
滑窗窗口长度：时间间隔, 滑动间隔: 时间间隔

withTumblingWindow(BaseWindowedBolt.Count count)
滚窗窗口长度：Tuple数

withTumblingWindow(BaseWindowedBolt.Duration duration)
滚窗窗口长度：时间间隔

以下代码是通过窗口操作来统计50s内每10s的Top-N，结合代码理解开窗操作。

github传送门：https://github.com/ZzzzZzreal/StormGoGo/blob/master/src/main/java/window/WindowTopN.java

public class WindowTopN {/*** 产生数据的Spout，随机生成指定word并发出*/static class MySpout extends BaseRichSpout {String[] words = {"aa","bb","cc","dd","ee","ff","gg"};Random random = new Random();SpoutOutputCollector collector;public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {this.collector=collector;}public void nextTuple() {Utils.sleep(500);collector.emit(new Values(words[random.nextInt(words.length)]));}public void declareOutputFields(OutputFieldsDeclarer declarer) {declarer.declare(new Fields("word"));}}/*** windowBolt，实现窗口操作，并统计指定时间内单位时间间隔内的Top3*/static class MyWindowBolt extends BaseWindowedBolt {//定义一个HashMap作wordcount用HashMap<String, Integer> hashMap = new HashMap<String, Integer>();public void execute(TupleWindow inputWindow) {//获取窗口内的内容List<Tuple> words = inputWindow.get();//wordcountfor (int i = 0; i < words.size(); i++) {String word = words.get(i).getString(0);Integer count = hashMap.get(word);if (count == null)count = 0;count++;hashMap.put(word, count);}//这里将map.entrySet()转换成listList<Map.Entry<String, Integer>> list = new ArrayList<Map.Entry<String, Integer>>(hashMap.entrySet());//然后通过比较器来实现排序Collections.sort(list, new Comparator<Map.Entry<String, Integer>>() {//升序排序public int compare(Map.Entry<String, Integer> o1,Map.Entry<String, Integer> o2) {return o2.getValue().compareTo(o1.getValue());}});//输出top3System.out.println("Top3:");for (int i = 0; i < 3; i++) {System.out.println("\t" + list.get(i).getKey() + ":" + list.get(i).getValue());}System.out.println("--------->");/**以下代码用于理解window的某些基础操作*/
//            List<Tuple> tuples = inputWindow.get();
//            List<Tuple> expired = inputWindow.getExpired();//获取到过期的tuple
//            List<Tuple> tuples = inputWindow.getNew();//获取到和上个窗口相比新加进去的tuple
//            System.out.println("滑动了一下！");
//            System.out.println(tuples.size());
//            System.out.println(expired.size());
//            for (Tuple tuple:tuples){
//
//                System.out.println(tuple.getValue(0));
//        }}public static void main(String[] args) {//构建拓扑TopologyBuilder topologyBuilder = new TopologyBuilder();topologyBuilder.setSpout("spout", new MySpout());//指定窗口长度以及滑动间隔topologyBuilder.setBolt("bolt", new MyWindowBolt().withWindow(BaseWindowedBolt.Duration.seconds(50), BaseWindowedBolt.Duration.seconds(10))).shuffleGrouping("spout");/**以下代码简单理解定义窗口时时间和数量的排列组合*/
//        topologyBuilder.setBolt("bolt", new MyWindowBolt().withTumblingWindow(BaseWindowedBolt.Count.of(10)))
//              .shuffleGrouping("spout");//这里要注意withTumblingWindow（滑动间隔和窗口长度是一样的）和withWindow的区别，如果忘了点进源码看一下（withWindow是一个tuple滑动一次）
//        topologyBuilder.setBolt("bolt",new new MyWindowBolt().withWindow(new BaseWindowedBolt.Duration(10, TimeUnit.SECONDS)))
//                new MyWindowBolt().withWindow(BaseWindowedBolt.Duration.seconds(50), BaseWindowedBolt.Duration.seconds(10)))//时间的两种定义方式
//                .shuffleGrouping("spout");LocalCluster localCluster = new LocalCluster();Config config = new Config();config.put(Config.TOPOLOGY_MESSAGE_TIMEOUT_SECS, 100);//要保证超时时间大于等于窗口长度+滑动间隔长度localCluster.submitTopology("a", config, topologyBuilder.createTopology());}}
}

Storm的窗口操作相关推荐

selenium之窗口操作
前言 webdriver经常会遇到在多个窗口之间切换的操作,例如在A窗口操作某个元素跳转到B窗口,然后又进入C窗口,此时如果需要进入A\B窗口操作就需要用到特殊的窗口处理方法窗口处理方法获取所有窗 ...
搞定vim的窗口操作
最近在给学生演示数据结构代码时,发现用一般的方法总会有不方便,如果使用ide又觉得太浪费了,后来觉得用vim就够了,使用buffer总会需要页面调来跳出,学生看起来容易迷糊.所以就研究了下vim的窗口 ...
python的窗口处理模块_Python tkinter模块弹出窗口及传值回到主窗口操作详解
本文实例讲述了Python tkinter模块弹出窗口及传值回到主窗口操作.分享给大家供大家参考,具体如下: 有些时候,我们需要使用弹出窗口,对程序的运行参数进行设置.有两种选择一.标准窗口如果只 ...
MySQL数据库安装-Windows10为例【超级详细安装过程(每一步的截图)、检测安装是否成功、登录数据库并查看版本号、将MySQL添加到环境变量中、在命令行窗口操作数据库】
哈哈,这是在哔哩哔哩网站上看的 MySQL安装步骤. 我安装的时候,截截图,希望对大家有所帮助. 链接:https://pan.baidu.com/s/1fKFQCVHMQOmbg84UyGrKB ...
Spark Streaming的窗口操作
2019独角兽企业重金招聘Python工程师标准>>> Spark Streaming的窗口操作博客分类: spark Spark Streaming的Window Operati ...
SparkStreaming - 窗口函数（窗口操作）
窗口操作就是把多个采集周期设置成一个窗口,一起来计算,然后进行滑动,根据设置的滑动大小. 窗口大小和滑动大小,要是采集周期的倍数 package date_10_17_SparkStreamingim ...
最大化窗口设置_Qt学习笔记4（窗口操作及插入图片）
窗口操作: Qt全屏显示函数 showFullScreen() Qt最大化显示函数 showMaximized() Qt最小化显示函数 showMinimized() Qt固定尺寸显示函数 resiz ...
py+seleneium(多窗口操作、alert窗口操作、元素、上传的一些方法)
在使用selenium时,会出现一些情况,就是有多个窗口的情况selenium怎么从一个窗口跳入到另外一个窗口操作,下面还是看代码 handle1 = driver.current_window_ha ...
Python中最全的窗口操作，如窗口最大化、最小化、窗口置顶、获取缩放比例等
Python窗口操作前言本文记录在Python中操作Windows应用窗口的操作. 这里的操作都是自己摸索+借助强大的搜索引擎整理出来的,我真棒!!! 知识点

Storm的窗口操作

Storm的窗口操作相关推荐

最新文章

热门文章