Slipstream中的无限滑动窗口（Infinite Window）

偏执的太偏执、 2021-09-18 02:20 523阅读 0赞

滑动窗口中有一个特例是无限滑动窗口——它的窗口长度为无限长（INFINITE）。它的意思是，窗口每滑动一个间隔会触发一次计算，但是每次触发计算的窗口都会包含所有之前的窗口覆盖的区间。例如：对于窗口长度为 INFINITE，滑动间隔为1s的滑动窗口，第一个窗口为\[0s, 1s)，第二个窗口为\[0s, 2s)，第三个窗口为\[0s, 3s)，依次类推。

**1、无限滑动窗口测试举例**

无限滑动窗口也可以分别使用系统时间和事件时间进行切分，这里以系统时间切分为例举例说明，事件时间切分与之类似。

*  创建一个输入流，用于接收Kafka传过来的数据

CREATE STREAM aaa (id STRING, letter STRING)
    
      ROW FORMAT DELIMITED FIELDS TERMINATED BY ','
    
      TBLPROPERTIES("topic"="yxy",
    
      "kafka.zookeeper"="node3:2181,node2:2181,node1:2181",
    
      "kafka.broker.list"="node3:9092,node2:9092,node1:9092",
    
      "transwarp.consumer.security.protocol"="SASL_PLAINTEXT",
    
      "transwarp.consumer.sasl.mechanism"="GSSAPI",
    
      "transwarp.consumer.sasl.kerberos.service.name"="kafka",
    
      "transwarp.consumer.sasl.jaas.config"="com.sun.security.auth.module.Krb5LoginModule required useKeyTab=true storeKey=true keyTab=\"/etc/slipstream1/conf/kafka.keytab\" principal=\"kafka@TDH\""
    
      );

*  创建一个Window Stream

create stream bbb as select * from aaa STREAMWINDOW w1 AS (LENGTH INFINITE SLIDE '30' SECOND);

流bbb为无限滑动窗口，每隔30秒统计一次数据。

*  创建表ccc，用于接收经过处理的流数据

CREATE TABLE ccc (letter STRING, sl STRING);

*  触发一个聚合操作，按照“letter”分组，统计不同letter出现的次数

INSERT into ccc SELECT letter, count(*) as sl FROM bbb GROUP BY letter;

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70][]

*  测试

启动Kafka生产者生产数据：

![20190219144321942.png][]

查询得到表ccc中的数据如下：

![20190219144321938.png][]

30秒之后再Kafka中再次输入数据：

![20190219144321944.png][]

查询得到表ccc中的数据如下：

![20190219144321943.png][]

**注意：**在使用Window Stream进行聚合的时候，最好先建一个Window Stream，然后再进行聚合。

**2、无限滑动窗口聚合存在的问题**

无限滑动窗口解决了对流传输过来所有数据无法进行聚合操作的问题，但是在一定时间内将聚合后的数据插入到表中时却不能实现覆盖原先表中的数据。在事件驱动模式（Slipstream有微批模式和事件驱动模式两种，TDH 5.0之后的版本默认开启事件驱动模式，可以在8180监控界面的ngmr.engine.mode参数项下设置参数，morphling为事件驱动模式，mapred为微批模式，推荐使用事件驱动模式）下，将存储结果数据的表建成es表或HBase表，因为其索引或rowkey的唯一性，可以实现数据的覆盖插入。

这里以es表为例进行说明。

同1中相同，在建表的时候将表建成es表，如下：

CREATE TABLE ccc (letter STRING, sl STRING) stored as ES with shard number 8 replication 1;

触发聚合操作，并开启Kafka生产者生产数据：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 1][]

查询表ccc可得：

![2019021914432227.png][]

继续生产数据：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 2][]

一段时间后查询表ccc可得：

![2019021914432245.png][]

通过建es表，实现了数据的覆盖插入。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70]: /images/20210726/08a668c2d5b146f4acc508d4028405ec.png
[20190219144321942.png]: /images/20210726/ff7e5da6a11840729e43e24859f92bcf.png
[20190219144321938.png]: /images/20210726/b3ba48f574274beba045a5f2ee11cf5d.png
[20190219144321944.png]: /images/20210726/250f0c8ce3894cb4a37b97b0b9745866.png
[20190219144321943.png]: /images/20210726/2db3ad3034c74bcea5f6843f166d7fc5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 1]: /images/20210726/94f643b85e6f4cc0852c59dcf475c2d2.png
[2019021914432227.png]: /images/20210726/5293903b50024075a94698a8ce5cd356.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 2]: /images/20210726/aca1817d5f1642a68bda5aa801bcdf8d.png
[2019021914432245.png]: /images/20210726/a67dd7fc01bc407d8fe9abb46c1227c4.png