如何在kSQL中SELECT时间戳最新的单条记录

Stream中每条记录都包含整数类型的CREATEDAT字段，取值示例为1657208128653，该Stream创建语句如下：

CREATE STREAM my_stream (NAME STRING, CREATEDAT INT) WITH (KAFKA_TOPIC = 'my_topic', PARTITIONS=2, VALUE_FORMAT = 'JSON');

需求为SELECT查询单条最新的记录。
执行如下命令时无任何结果输出：

SELECT name, count(*) as count FROM my_stream TIMESTAMP BY CREATEDAT GROUP BY name, TUMBLINGWINDOW(seconds,10);

待解决问题：

• 如何修正上述SELECT语句使其返回最新记录
• kSQL中是否支持按时间戳对记录进行排序

原语句无输出的原因 #

一共三个核心问题：

1. 字段类型不匹配：定义CREATEDAT为INT类型，但示例值1657208128653是13位毫秒级时间戳，ksql中INT为32位整型，最大值仅为2147483647（对应10位秒级时间戳），时间戳解析直接失败，无法正常处理记录。
2. 语法错误：滚动窗口的正确写法是TUMBLING(SIZE <数值> <时间单位>)，原语句中TUMBLINGWINDOW(seconds,10)函数名、参数格式均不合法，会直接抛出语法异常。
3. 逻辑不匹配+缺少输出参数：原语句是窗口聚合统计逻辑，和“取最新单条记录”的需求完全无关；同时ksql的持续流查询必须在末尾加EMIT CHANGES才会持续输出结果，缺少该参数时查询会一直挂起不返回内容。

修正步骤 #

1. 先修正Stream表结构，将CREATEDAT字段类型改为支持13位时间戳的BIGINT：

CREATE OR REPLACE STREAM my_stream (NAME STRING, CREATEDAT BIGINT) WITH (KAFKA_TOPIC = 'my_topic', PARTITIONS=2, VALUE_FORMAT = 'JSON');

2. 根据需求选择对应查询语句：

• 如果需要按name分组，取每个name对应的最新一条记录：

SELECT 
  name, 
  LATEST_BY_OFFSET(CREATEDAT) AS latest_created_time
FROM my_stream
TIMESTAMP BY CREATEDAT
GROUP BY name
EMIT CHANGES;

• 如果需要取全局最新的单条记录，可以通过虚拟分组字段实现：

SELECT 
  LATEST_BY_OFFSET(name) AS latest_name, 
  LATEST_BY_OFFSET(CREATEDAT) AS latest_created_time
FROM my_stream
TIMESTAMP BY CREATEDAT
GROUP BY 1
EMIT CHANGES;

kSQL是面向无限数据流的查询引擎，不支持传统关系型数据库的全局ORDER BY排序能力——因为流数据是持续流入无边界的，无法对全量数据做全局排序。
仅两种场景下可以实现按时间顺序输出：

• 有限历史数据排序：针对有限历史数据的推送查询，查询语句末尾加LIMIT参数限制返回条数（查询到指定条数后自动终止），此时可以支持基于时间戳的排序，仅对查询命中的有限历史数据生效。
• 同分区内顺序输出：通过TIMESTAMP BY指定事件时间列后，ksql会保证同分区内的记录按事件时间顺序处理输出，但跨分区的记录无法保证全局时间顺序。

内容的提问来源于stack exchange，提问作者alphanumeric