Storm Windowing storm滑动窗口简介

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

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

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

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来划分窗口:

Tuple落入到哪个窗口，是看tuple里的Event time。

窗口向后推进，主要依靠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的安装步骤

Kafka-Storm 集成部署

Storm在Ubuntu环境下的单机部署