在给定时间内只有一个输出的KSQL table group by

“id VARCHAR,src_ip VARCHAR,消息VARCHAR”

现在我需要查看失败的登录是否在给定时间内重复3次以上,然后发出警报。所以我创建了一个表,如下所示,

CREATE TABLE  231_console_failure AS \
      SELECT src_ip, count(*) \ 
      FROM console_failure \
      WINDOW TUMBLING (SIZE 30 SECONDS) \
      WHERE message = 'failed_login' \
      GROUP BY src_ip \
      HAVING count(*) > 3;

当有一个匹配,即30秒内超过3,它就会给出这个值。但是假设30秒内有10次尝试,那么消费者获取7条消息,每条消息的计数从4到10不等。

我知道我可以在脚本中处理这个问题,避免使用None,在给定的时间内只接受更高的计数。但是,有没有办法从上表创建一个流,该流只与KSQL中的groupby匹配消息?