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

可以用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，重新生成、运行，从命令行下输入数据。

*来自志愿者整理的flink邮件归档