你好,游客 登录 注册 搜索
背景:
阅读新闻

Storm Windowing storm滑动窗口简介

[日期:2017-01-25] 来源:Linux社区  作者:cutd [字体: ]

Storm Windowing

简介

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

  • 窗口的长度,可以是时间间隔或Tuple数量;
  • 滑动间隔(sliding Interval),可以是时间间隔或Tuple数量;

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

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)
    滚窗 窗口长度:时间间隔

Tuple时间戳和乱序

storm支持追踪源数据的时间戳。
Event time 和Process time
默认的时间戳是处理元组时的bolt窗口生成的,
Event time,事件时间,通常这个时间会带在Tuple中;
Process time,到某一个处理环节的时间。
举例:A今天早上9点告诉B,说C昨天晚上9点在滨江国际;
这条信息中,可以认为C在滨江国际的Event time是昨天晚上9点,B接收到这条信息的时间,即Process time,是今天早上9点。

配置时间戳字段(timestamp field)

windows按照时间划分时,默认是Process time,也可以指定为Tuple中的Event time。
如果以Event time来划分窗口:

  1. Tuple落入到哪个窗口,是看tuple里的Event time。
  2. 窗口向后推进,主要依靠Event time的增长;
public BaseWindowedBolt withTimestampField(String fieldName)

延时(lag)和水位线(watermark)

从当前最后一条数据算起,往前减去lag,得到一个时间,这个时间就是watermark;
认为watermark之前的数据都已经到了。收到06:01:00的数据时,认为06:00:00的数据都到了。给他们入window。
这样实际是一个延时处理,等到了06:01:00时,我才开始将06:00:00的数据放入窗口。

如果很不巧,06:00:00的数据在06:01:00之后,lag为60s,不好意思,进不了窗口。此数据不会被处理,并且会在worker的日志中加一行INFO信息。

public class SlidingWindowBolt extends BaseWindowedBolt {
    private OutputCollector collector;

    @Override
    publicvoidprepare(Map stormConf, TopologyContext context, OutputCollector collector) {
        this.collector = collector;
    }

    @Override
    publicvoidexecute(TupleWindow inputWindow) {
      for(Tuple tuple: inputWindow.get()) {
        // do the windowing computation
        ...
      }
      // emit the results
      collector.emit(new Values(computedValue));
    }
}

publicstaticvoidmain(String[] args) {
    TopologyBuilder builder = new TopologyBuilder();
     builder.setSpout("spout", new RandomSentenceSpout(), 1);
     builder.setBolt("slidingwindowbolt",
                     new SlidingWindowBolt().withWindow(new Count(30), new Count(10)),
                     1).shuffleGrouping("spout");
    Config conf = new Config();
    conf.setDebug(true);
    conf.setNumWorkers(1);

    StormSubmitter.submitTopologyWithProgressBar(args[0], conf, builder.createTopology());

}

Storm的安装步骤 http://www.linuxidc.com/Linux/2016-08/134184.htm

Kafka-Storm 集成部署 http://www.linuxidc.com/Linux/2016-03/129063.htm

Storm在Ubuntu环境下的单机部署 http://www.linuxidc.com/Linux/2016-03/129060.htm

本文永久更新链接地址http://www.linuxidc.com/Linux/2017-01/139946.htm

linux
相关资讯       Storm 
本文评论   查看全部评论 (0)
表情: 表情 姓名: 字数

       

评论声明
  • 尊重网上道德,遵守中华人民共和国的各项有关法律法规
  • 承担一切因您的行为而直接或间接导致的民事或刑事法律责任
  • 本站管理人员有权保留或删除其管辖留言中的任意内容
  • 本站有权在网站内转载或引用您的评论
  • 参与本评论即表明您已经阅读并接受上述条款