apache-kafka - KSQL UNIX_TIMESTAMP 函数在使用查询创建的流上不是动态的

Question

我正在创建以下流：

CREATE STREAM riderLocations (profileId VARCHAR, latitude DOUBLE, longitude DOUBLE, publishtime VARCHAR) 
WITH (kafka_topic='locations', value_format='json', partitions=1);

然后是另一个：

CREATE STREAM INVENTORY WITH (KAFKA_TOPIC='locations_in') 
AS select * FROM riderLocations 
where STRINGTOTIMESTAMP(publishtime, 'yyyy-MM-dd HH:mm:ss.SSSZ') < UNIX_TIMESTAMP();

当我执行命令时：select * from inventory emit changes;

它仅显示发布日期小于创建库存流时的消息。

如何强制 unix_timestamp 值更新和更新我的流库存？

标签： apache-kafkaksqldb

Answer 1

时间戳是在查询创建时评估的，而不是运行时。

如果您希望消息少于“现在”，则需要将其作为您运行的每个 SELECT 查询的 WHERE 子句包含在内