如果用flink sql持续查询过去30分钟登录网站的人数?

classic Classic list List threaded Threaded
6 messages Options
Reply | Threaded
Open this post in threaded view
|

如果用flink sql持续查询过去30分钟登录网站的人数?

陈帅
例如,用户在以下时间点登录:无, 12:02, 12:13, 12:15, 12:31, 12:40, 12:45, 无
那么我期望在以下时间点(实际查询可能在任意时间点)获取到的结果数为
12:01 (0),  12:03:(1),  12:14 (2),  12:16(3), 12:30 (4), 12:35 (4), 12:41
(5), 12:46 (4), 13:16 (0)
即每个元素进来就会设一个30分钟过期时间,窗口状态是维护还当前未过期元素集合。

如果用sliding window的话,步长需要设置成1秒,那么窗口个数会膨胀很多,而实际上我只需要统计其中一个窗口,多余的窗口浪费了。我也考虑过用
over window,但是不知道它是否支持处理时间,因为我的场景是需要根据处理时间推移而改变统计值的。我尝试用stream
api来实现,利用了timerService设置元素过期时间,但我测下来发现元素过期速度赶不上进入元素的速度,导致state大小一直增长.

所以想问一下:
1. 针对这种case有没有标准做法?sql支持吗?
2. 要怎么解决timerService的性能问题?timerService底层实现是不是单线程处理priority queue?

谢谢!
陈帅
Reply | Threaded
Open this post in threaded view
|

回复: 如果用flink sql持续查询过去30分钟登录网站的人数?

Yuan,Youjun
可以用30分钟的range over窗口来处理,但是你提到两个0值得输出恐怕做不到,没有数据,没有产出。
假设你得输入包含ts和userid两个字段,分别为时间戳和用户id,那么SQL应该这样:
INSERT INTO mysink
SELECT
   ts, userid,  
   COUNT(userid)  
       OVER (PARTITION BY userid ORDER BY rowtime RANGE BETWEEN INTERVAL '30' MINUTE PRECEDING AND CURRENT ROW) AS cnt
FROM mysrc

以如下输入为例:
"2019-12-05 12:02:00,user1",
"2019-12-05 12:13:00,user1",
"2019-12-05 12:15:00,user1",
"2019-12-05 12:31:00,user1",
"2019-12-05 12:40:00,user1",
"2019-12-05 12:45:00,user1"
产出如下结果:
{"cnt":1,"ts":1575547320000,"userid":"user1"}
{"cnt":2,"ts":1575547980000,"userid":"user1"}
{"cnt":3,"ts":1575548100000,"userid":"user1"}
{"cnt":4,"ts":1575549060000,"userid":"user1"}
{"cnt":4,"ts":1575549600000,"userid":"user1"}
{"cnt":4,"ts":1575549900000,"userid":"user1"}

为了验证上述SQL,你可以将如下作业粘贴到http://creek.baidubce.com/ 的作业定义输入框中,点击生成可执行文件,运行下载到的可执行文件,就能看到结果:
{
    "sources": [{
        "schema": {
            "format": "CSV",
            "fields": [{
                    "name": "ts",
                    "type": "SQL_TIMESTAMP"
                },
                {
                    "name": "userid",
                    "type": "STRING"
                }]
        },
        "watermark": 0,
        "name": "mysrc",
        "eventTime": "ts",
        "type": "COLLECTION",
      "attr": {
      "input":[
          "2019-12-05 12:02:00,user1",
            "2019-12-05 12:13:00,user1",
                        "2019-12-05 12:15:00,user1",
                        "2019-12-05 12:31:00,user1",
                        "2019-12-05 12:40:00,user1",
                        "2019-12-05 12:45:00,user1"
          ]
      }
    }],
    "sink": {
        "schema": {
            "format": "JSON"
        },
        "name": "mysink",
        "type": "STDOUT"
    },
    "name": "demojob",
    "timeType": "EVENTTIME",
    "sql": "INSERT INTO mysink SELECT rowtime, userid, COUNT(userid)  OVER (PARTITION BY userid ORDER BY rowtime RANGE BETWEEN INTERVAL '30' MINUTE PRECEDING AND CURRENT ROW) AS cnt FROM mysrc"
}

当然上面的例子是以事件时间,用处理时间也是可以的。为了验证,你可以把source.type从COLLECTION改成STDIN,把timeType从EVENTTIME改成PROCESSTIME,重新生成、运行,从命令行下输入数据。

袁尤军

-----邮件原件-----
发件人: 陈帅 <[hidden email]>
发送时间: Wednesday, December 4, 2019 11:40 PM
收件人: [hidden email]
主题: 如果用flink sql持续查询过去30分钟登录网站的人数?

例如,用户在以下时间点登录:无, 12:02, 12:13, 12:15, 12:31, 12:40, 12:45, 无
那么我期望在以下时间点(实际查询可能在任意时间点)获取到的结果数为
12:01 (0),  12:03:(1),  12:14 (2),  12:16(3), 12:30 (4), 12:35 (4), 12:41 (5), 12:46 (4), 13:16 (0)
即每个元素进来就会设一个30分钟过期时间,窗口状态是维护还当前未过期元素集合。

如果用sliding window的话,步长需要设置成1秒,那么窗口个数会膨胀很多,而实际上我只需要统计其中一个窗口,多余的窗口浪费了。我也考虑过用
over window,但是不知道它是否支持处理时间,因为我的场景是需要根据处理时间推移而改变统计值的。我尝试用stream
api来实现,利用了timerService设置元素过期时间,但我测下来发现元素过期速度赶不上进入元素的速度,导致state大小一直增长.

所以想问一下:
1. 针对这种case有没有标准做法?sql支持吗?
2. 要怎么解决timerService的性能问题?timerService底层实现是不是单线程处理priority queue?

谢谢!
陈帅
Reply | Threaded
Open this post in threaded view
|

Re: 如果用flink sql持续查询过去30分钟登录网站的人数?

陈帅
你们这个平台还挺方便快速验证的,是扩展了Flink SQL吗?
虽然没有完全解决我的问题,但还是要谢谢你。

Yuan,Youjun <[hidden email]> 于2019年12月5日周四 上午10:41写道:

> 可以用30分钟的range over窗口来处理,但是你提到两个0值得输出恐怕做不到,没有数据,没有产出。
> 假设你得输入包含ts和userid两个字段,分别为时间戳和用户id,那么SQL应该这样:
> INSERT INTO mysink
> SELECT
>    ts, userid,
>    COUNT(userid)
>        OVER (PARTITION BY userid ORDER BY rowtime RANGE BETWEEN INTERVAL
> '30' MINUTE PRECEDING AND CURRENT ROW) AS cnt
> FROM mysrc
>
> 以如下输入为例:
> "2019-12-05 12:02:00,user1",
> "2019-12-05 12:13:00,user1",
> "2019-12-05 12:15:00,user1",
> "2019-12-05 12:31:00,user1",
> "2019-12-05 12:40:00,user1",
> "2019-12-05 12:45:00,user1"
> 产出如下结果:
> {"cnt":1,"ts":1575547320000,"userid":"user1"}
> {"cnt":2,"ts":1575547980000,"userid":"user1"}
> {"cnt":3,"ts":1575548100000,"userid":"user1"}
> {"cnt":4,"ts":1575549060000,"userid":"user1"}
> {"cnt":4,"ts":1575549600000,"userid":"user1"}
> {"cnt":4,"ts":1575549900000,"userid":"user1"}
>
> 为了验证上述SQL,你可以将如下作业粘贴到http://creek.baidubce.com/
> 的作业定义输入框中,点击生成可执行文件,运行下载到的可执行文件,就能看到结果:
> {
>     "sources": [{
>         "schema": {
>             "format": "CSV",
>             "fields": [{
>                     "name": "ts",
>                     "type": "SQL_TIMESTAMP"
>                 },
>                 {
>                     "name": "userid",
>                     "type": "STRING"
>                 }]
>         },
>         "watermark": 0,
>         "name": "mysrc",
>         "eventTime": "ts",
>         "type": "COLLECTION",
>         "attr": {
>                 "input":[
>                 "2019-12-05 12:02:00,user1",
>             "2019-12-05 12:13:00,user1",
>                         "2019-12-05 12:15:00,user1",
>                         "2019-12-05 12:31:00,user1",
>                         "2019-12-05 12:40:00,user1",
>                         "2019-12-05 12:45:00,user1"
>           ]
>       }
>     }],
>     "sink": {
>         "schema": {
>             "format": "JSON"
>         },
>         "name": "mysink",
>         "type": "STDOUT"
>     },
>     "name": "demojob",
>     "timeType": "EVENTTIME",
>     "sql": "INSERT INTO mysink SELECT rowtime, userid, COUNT(userid)  OVER
> (PARTITION BY userid ORDER BY rowtime RANGE BETWEEN INTERVAL '30' MINUTE
> PRECEDING AND CURRENT ROW) AS cnt FROM mysrc"
> }
>
>
> 当然上面的例子是以事件时间,用处理时间也是可以的。为了验证,你可以把source.type从COLLECTION改成STDIN,把timeType从EVENTTIME改成PROCESSTIME,重新生成、运行,从命令行下输入数据。
>
> 袁尤军
>
> -----邮件原件-----
> 发件人: 陈帅 <[hidden email]>
> 发送时间: Wednesday, December 4, 2019 11:40 PM
> 收件人: [hidden email]
> 主题: 如果用flink sql持续查询过去30分钟登录网站的人数?
>
> 例如,用户在以下时间点登录:无, 12:02, 12:13, 12:15, 12:31, 12:40, 12:45, 无
> 那么我期望在以下时间点(实际查询可能在任意时间点)获取到的结果数为
> 12:01 (0),  12:03:(1),  12:14 (2),  12:16(3), 12:30 (4), 12:35 (4), 12:41
> (5), 12:46 (4), 13:16 (0)
> 即每个元素进来就会设一个30分钟过期时间,窗口状态是维护还当前未过期元素集合。
>
> 如果用sliding window的话,步长需要设置成1秒,那么窗口个数会膨胀很多,而实际上我只需要统计其中一个窗口,多余的窗口浪费了。我也考虑过用
> over window,但是不知道它是否支持处理时间,因为我的场景是需要根据处理时间推移而改变统计值的。我尝试用stream
> api来实现,利用了timerService设置元素过期时间,但我测下来发现元素过期速度赶不上进入元素的速度,导致state大小一直增长.
>
> 所以想问一下:
> 1. 针对这种case有没有标准做法?sql支持吗?
> 2. 要怎么解决timerService的性能问题?timerService底层实现是不是单线程处理priority queue?
>
> 谢谢!
> 陈帅
>
Reply | Threaded
Open this post in threaded view
|

回复: 如果用flink sql持续查询过去30分钟登录网站的人数?

Yuan,Youjun
是的,与Flink完全一样的SQL接口,为边缘计算设计的流式计算框架。


-----邮件原件-----
发件人: 陈帅 <[hidden email]>
发送时间: Saturday, December 7, 2019 11:36 AM
收件人: [hidden email]
主题: Re: 如果用flink sql持续查询过去30分钟登录网站的人数?

你们这个平台还挺方便快速验证的,是扩展了Flink SQL吗?
虽然没有完全解决我的问题,但还是要谢谢你。

Yuan,Youjun <[hidden email]> 于2019年12月5日周四 上午10:41写道:

> 可以用30分钟的range over窗口来处理,但是你提到两个0值得输出恐怕做不到,没有数据,没有产出。
> 假设你得输入包含ts和userid两个字段,分别为时间戳和用户id,那么SQL应该这样:
> INSERT INTO mysink
> SELECT
>    ts, userid,
>    COUNT(userid)
>        OVER (PARTITION BY userid ORDER BY rowtime RANGE BETWEEN
> INTERVAL '30' MINUTE PRECEDING AND CURRENT ROW) AS cnt FROM mysrc
>
> 以如下输入为例:
> "2019-12-05 12:02:00,user1",
> "2019-12-05 12:13:00,user1",
> "2019-12-05 12:15:00,user1",
> "2019-12-05 12:31:00,user1",
> "2019-12-05 12:40:00,user1",
> "2019-12-05 12:45:00,user1"
> 产出如下结果:
> {"cnt":1,"ts":1575547320000,"userid":"user1"}
> {"cnt":2,"ts":1575547980000,"userid":"user1"}
> {"cnt":3,"ts":1575548100000,"userid":"user1"}
> {"cnt":4,"ts":1575549060000,"userid":"user1"}
> {"cnt":4,"ts":1575549600000,"userid":"user1"}
> {"cnt":4,"ts":1575549900000,"userid":"user1"}
>
> 为了验证上述SQL,你可以将如下作业粘贴到http://creek.baidubce.com/
> 的作业定义输入框中,点击生成可执行文件,运行下载到的可执行文件,就能看到结果:
> {
>     "sources": [{
>         "schema": {
>             "format": "CSV",
>             "fields": [{
>                     "name": "ts",
>                     "type": "SQL_TIMESTAMP"
>                 },
>                 {
>                     "name": "userid",
>                     "type": "STRING"
>                 }]
>         },
>         "watermark": 0,
>         "name": "mysrc",
>         "eventTime": "ts",
>         "type": "COLLECTION",
>         "attr": {
>                 "input":[
>                 "2019-12-05 12:02:00,user1",
>             "2019-12-05 12:13:00,user1",
>                         "2019-12-05 12:15:00,user1",
>                         "2019-12-05 12:31:00,user1",
>                         "2019-12-05 12:40:00,user1",
>                         "2019-12-05 12:45:00,user1"
>           ]
>       }
>     }],
>     "sink": {
>         "schema": {
>             "format": "JSON"
>         },
>         "name": "mysink",
>         "type": "STDOUT"
>     },
>     "name": "demojob",
>     "timeType": "EVENTTIME",
>     "sql": "INSERT INTO mysink SELECT rowtime, userid, COUNT(userid)  
> OVER (PARTITION BY userid ORDER BY rowtime RANGE BETWEEN INTERVAL '30'
> MINUTE PRECEDING AND CURRENT ROW) AS cnt FROM mysrc"
> }
>
>
> 当然上面的例子是以事件时间,用处理时间也是可以的。为了验证,你可以把source.type从COLLECTION改成STDIN,把timeT
> ype从EVENTTIME改成PROCESSTIME,重新生成、运行,从命令行下输入数据。
>
> 袁尤军
>
> -----邮件原件-----
> 发件人: 陈帅 <[hidden email]>
> 发送时间: Wednesday, December 4, 2019 11:40 PM
> 收件人: [hidden email]
> 主题: 如果用flink sql持续查询过去30分钟登录网站的人数?
>
> 例如,用户在以下时间点登录:无, 12:02, 12:13, 12:15, 12:31, 12:40, 12:45, 无
> 那么我期望在以下时间点(实际查询可能在任意时间点)获取到的结果数为
> 12:01 (0),  12:03:(1),  12:14 (2),  12:16(3), 12:30 (4), 12:35 (4),
> 12:41 (5), 12:46 (4), 13:16 (0)
> 即每个元素进来就会设一个30分钟过期时间,窗口状态是维护还当前未过期元素集合。
>
> 如果用sliding
> window的话,步长需要设置成1秒,那么窗口个数会膨胀很多,而实际上我只需要统计其中一个窗口,多余的窗口浪费了。我也考虑过用
> over window,但是不知道它是否支持处理时间,因为我的场景是需要根据处理时间推移而改变统计值的。我尝试用stream
> api来实现,利用了timerService设置元素过期时间,但我测下来发现元素过期速度赶不上进入元素的速度,导致state大小一直增长.
>
> 所以想问一下:
> 1. 针对这种case有没有标准做法?sql支持吗?
> 2. 要怎么解决timerService的性能问题?timerService底层实现是不是单线程处理priority queue?
>
> 谢谢!
> 陈帅
>
Reply | Threaded
Open this post in threaded view
|

Re: 如果用flink sql持续查询过去30分钟登录网站的人数?

陈帅
请问这个平台生成可执行文件creek是如何实现的?对应的Dawin-amd64环境下载下来的文件是什么格式的?

Yuan,Youjun <[hidden email]> 于2019年12月7日周六 下午8:32写道:

> 是的,与Flink完全一样的SQL接口,为边缘计算设计的流式计算框架。
>
>
> -----邮件原件-----
> 发件人: 陈帅 <[hidden email]>
> 发送时间: Saturday, December 7, 2019 11:36 AM
> 收件人: [hidden email]
> 主题: Re: 如果用flink sql持续查询过去30分钟登录网站的人数?
>
> 你们这个平台还挺方便快速验证的,是扩展了Flink SQL吗?
> 虽然没有完全解决我的问题,但还是要谢谢你。
>
> Yuan,Youjun <[hidden email]> 于2019年12月5日周四 上午10:41写道:
>
> > 可以用30分钟的range over窗口来处理,但是你提到两个0值得输出恐怕做不到,没有数据,没有产出。
> > 假设你得输入包含ts和userid两个字段,分别为时间戳和用户id,那么SQL应该这样:
> > INSERT INTO mysink
> > SELECT
> >    ts, userid,
> >    COUNT(userid)
> >        OVER (PARTITION BY userid ORDER BY rowtime RANGE BETWEEN
> > INTERVAL '30' MINUTE PRECEDING AND CURRENT ROW) AS cnt FROM mysrc
> >
> > 以如下输入为例:
> > "2019-12-05 12:02:00,user1",
> > "2019-12-05 12:13:00,user1",
> > "2019-12-05 12:15:00,user1",
> > "2019-12-05 12:31:00,user1",
> > "2019-12-05 12:40:00,user1",
> > "2019-12-05 12:45:00,user1"
> > 产出如下结果:
> > {"cnt":1,"ts":1575547320000,"userid":"user1"}
> > {"cnt":2,"ts":1575547980000,"userid":"user1"}
> > {"cnt":3,"ts":1575548100000,"userid":"user1"}
> > {"cnt":4,"ts":1575549060000,"userid":"user1"}
> > {"cnt":4,"ts":1575549600000,"userid":"user1"}
> > {"cnt":4,"ts":1575549900000,"userid":"user1"}
> >
> > 为了验证上述SQL,你可以将如下作业粘贴到http://creek.baidubce.com/
> > 的作业定义输入框中,点击生成可执行文件,运行下载到的可执行文件,就能看到结果:
> > {
> >     "sources": [{
> >         "schema": {
> >             "format": "CSV",
> >             "fields": [{
> >                     "name": "ts",
> >                     "type": "SQL_TIMESTAMP"
> >                 },
> >                 {
> >                     "name": "userid",
> >                     "type": "STRING"
> >                 }]
> >         },
> >         "watermark": 0,
> >         "name": "mysrc",
> >         "eventTime": "ts",
> >         "type": "COLLECTION",
> >         "attr": {
> >                 "input":[
> >                 "2019-12-05 12:02:00,user1",
> >             "2019-12-05 12:13:00,user1",
> >                         "2019-12-05 12:15:00,user1",
> >                         "2019-12-05 12:31:00,user1",
> >                         "2019-12-05 12:40:00,user1",
> >                         "2019-12-05 12:45:00,user1"
> >           ]
> >       }
> >     }],
> >     "sink": {
> >         "schema": {
> >             "format": "JSON"
> >         },
> >         "name": "mysink",
> >         "type": "STDOUT"
> >     },
> >     "name": "demojob",
> >     "timeType": "EVENTTIME",
> >     "sql": "INSERT INTO mysink SELECT rowtime, userid, COUNT(userid)
> > OVER (PARTITION BY userid ORDER BY rowtime RANGE BETWEEN INTERVAL '30'
> > MINUTE PRECEDING AND CURRENT ROW) AS cnt FROM mysrc"
> > }
> >
> >
> > 当然上面的例子是以事件时间,用处理时间也是可以的。为了验证,你可以把source.type从COLLECTION改成STDIN,把timeT
> > ype从EVENTTIME改成PROCESSTIME,重新生成、运行,从命令行下输入数据。
> >
> > 袁尤军
> >
> > -----邮件原件-----
> > 发件人: 陈帅 <[hidden email]>
> > 发送时间: Wednesday, December 4, 2019 11:40 PM
> > 收件人: [hidden email]
> > 主题: 如果用flink sql持续查询过去30分钟登录网站的人数?
> >
> > 例如,用户在以下时间点登录:无, 12:02, 12:13, 12:15, 12:31, 12:40, 12:45, 无
> > 那么我期望在以下时间点(实际查询可能在任意时间点)获取到的结果数为
> > 12:01 (0),  12:03:(1),  12:14 (2),  12:16(3), 12:30 (4), 12:35 (4),
> > 12:41 (5), 12:46 (4), 13:16 (0)
> > 即每个元素进来就会设一个30分钟过期时间,窗口状态是维护还当前未过期元素集合。
> >
> > 如果用sliding
> > window的话,步长需要设置成1秒,那么窗口个数会膨胀很多,而实际上我只需要统计其中一个窗口,多余的窗口浪费了。我也考虑过用
> > over window,但是不知道它是否支持处理时间,因为我的场景是需要根据处理时间推移而改变统计值的。我尝试用stream
> > api来实现,利用了timerService设置元素过期时间,但我测下来发现元素过期速度赶不上进入元素的速度,导致state大小一直增长.
> >
> > 所以想问一下:
> > 1. 针对这种case有没有标准做法?sql支持吗?
> > 2. 要怎么解决timerService的性能问题?timerService底层实现是不是单线程处理priority queue?
> >
> > 谢谢!
> > 陈帅
> >
>
Reply | Threaded
Open this post in threaded view
|

回复: 如果用flink sql持续查询过去30分钟登录网站的人数?

Yuan,Youjun
不好意思,之前没看到这个问题。
Darwin-amd64就是mac上的可执行文件格式。信任他,直接可以执行的。

-----邮件原件-----
发件人: 陈帅 <[hidden email]>
发送时间: Saturday, December 7, 2019 10:48 PM
收件人: [hidden email]
主题: Re: 如果用flink sql持续查询过去30分钟登录网站的人数?

请问这个平台生成可执行文件creek是如何实现的?对应的Dawin-amd64环境下载下来的文件是什么格式的?

Yuan,Youjun <[hidden email]> 于2019年12月7日周六 下午8:32写道:

> 是的,与Flink完全一样的SQL接口,为边缘计算设计的流式计算框架。
>
>
> -----邮件原件-----
> 发件人: 陈帅 <[hidden email]>
> 发送时间: Saturday, December 7, 2019 11:36 AM
> 收件人: [hidden email]
> 主题: Re: 如果用flink sql持续查询过去30分钟登录网站的人数?
>
> 你们这个平台还挺方便快速验证的,是扩展了Flink SQL吗?
> 虽然没有完全解决我的问题,但还是要谢谢你。
>
> Yuan,Youjun <[hidden email]> 于2019年12月5日周四 上午10:41写道:
>
> > 可以用30分钟的range over窗口来处理,但是你提到两个0值得输出恐怕做不到,没有数据,没有产出。
> > 假设你得输入包含ts和userid两个字段,分别为时间戳和用户id,那么SQL应该这样:
> > INSERT INTO mysink
> > SELECT
> >    ts, userid,
> >    COUNT(userid)
> >        OVER (PARTITION BY userid ORDER BY rowtime RANGE BETWEEN
> > INTERVAL '30' MINUTE PRECEDING AND CURRENT ROW) AS cnt FROM mysrc
> >
> > 以如下输入为例:
> > "2019-12-05 12:02:00,user1",
> > "2019-12-05 12:13:00,user1",
> > "2019-12-05 12:15:00,user1",
> > "2019-12-05 12:31:00,user1",
> > "2019-12-05 12:40:00,user1",
> > "2019-12-05 12:45:00,user1"
> > 产出如下结果:
> > {"cnt":1,"ts":1575547320000,"userid":"user1"}
> > {"cnt":2,"ts":1575547980000,"userid":"user1"}
> > {"cnt":3,"ts":1575548100000,"userid":"user1"}
> > {"cnt":4,"ts":1575549060000,"userid":"user1"}
> > {"cnt":4,"ts":1575549600000,"userid":"user1"}
> > {"cnt":4,"ts":1575549900000,"userid":"user1"}
> >
> > 为了验证上述SQL,你可以将如下作业粘贴到http://creek.baidubce.com/
> > 的作业定义输入框中,点击生成可执行文件,运行下载到的可执行文件,就能看到结果:
> > {
> >     "sources": [{
> >         "schema": {
> >             "format": "CSV",
> >             "fields": [{
> >                     "name": "ts",
> >                     "type": "SQL_TIMESTAMP"
> >                 },
> >                 {
> >                     "name": "userid",
> >                     "type": "STRING"
> >                 }]
> >         },
> >         "watermark": 0,
> >         "name": "mysrc",
> >         "eventTime": "ts",
> >         "type": "COLLECTION",
> >         "attr": {
> >                 "input":[
> >                 "2019-12-05 12:02:00,user1",
> >             "2019-12-05 12:13:00,user1",
> >                         "2019-12-05 12:15:00,user1",
> >                         "2019-12-05 12:31:00,user1",
> >                         "2019-12-05 12:40:00,user1",
> >                         "2019-12-05 12:45:00,user1"
> >           ]
> >       }
> >     }],
> >     "sink": {
> >         "schema": {
> >             "format": "JSON"
> >         },
> >         "name": "mysink",
> >         "type": "STDOUT"
> >     },
> >     "name": "demojob",
> >     "timeType": "EVENTTIME",
> >     "sql": "INSERT INTO mysink SELECT rowtime, userid, COUNT(userid)
> > OVER (PARTITION BY userid ORDER BY rowtime RANGE BETWEEN INTERVAL '30'
> > MINUTE PRECEDING AND CURRENT ROW) AS cnt FROM mysrc"
> > }
> >
> >
> > 当然上面的例子是以事件时间,用处理时间也是可以的。为了验证,你可以把source.type从COLLECTION改成STDIN,把tim
> > eT
> > ype从EVENTTIME改成PROCESSTIME,重新生成、运行,从命令行下输入数据。
> >
> > 袁尤军
> >
> > -----邮件原件-----
> > 发件人: 陈帅 <[hidden email]>
> > 发送时间: Wednesday, December 4, 2019 11:40 PM
> > 收件人: [hidden email]
> > 主题: 如果用flink sql持续查询过去30分钟登录网站的人数?
> >
> > 例如,用户在以下时间点登录:无, 12:02, 12:13, 12:15, 12:31, 12:40, 12:45, 无
> > 那么我期望在以下时间点(实际查询可能在任意时间点)获取到的结果数为
> > 12:01 (0),  12:03:(1),  12:14 (2),  12:16(3), 12:30 (4), 12:35 (4),
> > 12:41 (5), 12:46 (4), 13:16 (0)
> > 即每个元素进来就会设一个30分钟过期时间,窗口状态是维护还当前未过期元素集合。
> >
> > 如果用sliding
> > window的话,步长需要设置成1秒,那么窗口个数会膨胀很多,而实际上我只需要统计其中一个窗口,多余的窗口浪费了。我也考虑过用
> > over window,但是不知道它是否支持处理时间,因为我的场景是需要根据处理时间推移而改变统计值的。我尝试用stream
> > api来实现,利用了timerService设置元素过期时间,但我测下来发现元素过期速度赶不上进入元素的速度,导致state大小一直增长.
> >
> > 所以想问一下:
> > 1. 针对这种case有没有标准做法?sql支持吗?
> > 2. 要怎么解决timerService的性能问题?timerService底层实现是不是单线程处理priority queue?
> >
> > 谢谢!
> > 陈帅
> >
>