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值更新和更新我的流库存？

Answer 1

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

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

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

KSQL UNIX_TIMESTAMP function is not dinamic on streams created with queries

apache-kafka

ksqldb