KSQL窗口聚合流

Question

我有一个流，它是由一个kafka主题创建的，并且指定了Timestamp属性。

当我试图创建一个流时，会话窗口化了一个查询，如下所示：

CREATE STREAM SESSION_STREAM AS
SELECT ...
  FROM EVENT_STREAM
WINDOW SESSION (5 MINUTES)
   GROUP BY ...;

我总是得到错误：

CREATE STREAM SESSION_START_STREAM AS
SELECT *
  FROM SESSION_TABLE
 WHERE WINDOWSTART=WINDOWEND;

KSQL不支持对窗口表的持久查询

如何在KSQL中创建开始会话窗口的事件的流？

Answer 1

如果将create stream语句切换为create table语句，则将创建一个不断更新的表。接收器主题session_stream将包含对表的更改流，即它的changelog。

ksqlDB将其建模为表，因为它具有表语义，即表中只能存在具有任何特定键的单行。但是，changelog将包含已应用于表的更改流。

如果您想要的是一个包含所有会话的主题，那么类似这样的内容将创建：

-- create a stream with a new 'data' topic:
CREATE STREAM DATA (USER_ID INT) 
    WITH (kafka_topic='data', value_format='json');

-- create a table that tracks user interactions per session:
CREATE TABLE SESSION AS
SELECT USER_ID, COUNT(USER_ID) AS COUNT
  FROM DATA
WINDOW SESSION (5 SECONDS)
   GROUP BY USER_ID;

-- Create a stream over the existing `SESSIONS` topic.
-- Note it states the window_type is 'Session'.
CREATE STREAM SESSION_STREAM (ROWKEY INT KEY, COUNT BIGINT) 
   WITH (kafka_topic='SESSIONS', value_format='JSON', window_type='Session');

-- Create a stream of window start events:
CREATE STREAM SESSION_STARTS AS 
    SELECT * FROM SESSION_STREAM 
    WHERE WINDOWSTART = WINDOWEND;

CREATE STREAM SESSION_STREAM (USER_ID INT KEY, COUNT BIGINT) 
   WITH (kafka_topic='SESSIONS', value_format='JSON', window_type='Session');