KSQL 窗口化聚合流

Question

我正在尝试使用KSQL Windowed Aggregation，特别是Session Window，按其属性之一对事件进行分组。

我有一个 STREAM 由 kafka 主题制成，并明确指定了 TIMESTAMP 属性。

当我尝试使用会话窗口创建STREAM 时，查询如下：

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

我总是得到错误：

您的 SELECT 查询会生成一个 TABLE。请改用 CREATE TABLE AS SELECT 语句。

是否可以使用窗口聚合创建 STREAM？

---

当我按照建议尝试创建一个 TABLE 然后创建一个包含所有会话开始事件的 STREAM 时，查询如下：

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

KSQL 告诉我：

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

如何在 KSQL 中创建一个 STREAM 事件启动会话窗口？

Answer 1

如果切换到 create table 语句，您的 create stream 语句将创建一个不断更新的表。接收器主题SESSION_STREAM 将包含对表的更改流，即其更改日志。

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

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

-- 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;

这将创建一个 SESSIONS 主题，其中包含对 SESSIONS 表的更改：即其更改日志。

如果您想将其转换为会话开始事件流，那么不幸的是，ksqlDB还不允许您直接更改从表中创建流，但您可以通过以下方式创建流表的更改日志：

-- 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;

注意，随着即将发布的 0.10 版本，您将能够正确命名 SESSION_STREAM 中的键列：

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