在KsqlDB中使用WHERE子句创建新的筛选流不起作用

inkz8wg9 于 2022-09-21 发布在 Apache

关注(0)|答案(2)|浏览(138)

因此，我从一个主题创建了一个流。然后我想我应该使用WHERE子句从那里创建一个经过过滤的流。

有点像这样：

CREATE STREAM filtered as select * from original where property = 'value';

但是，当我从该流中进行选择时，所有条目都不会被过滤。当我在原始流上运行带有WHERE子句的查询时，就像一个普通的SELECT PULL查询一样，它按预期工作。我还可以对“筛选的”查询运行相同的SELECT/WHERE查询并获得筛选的结果，但这不是本练习的重点。

apache-kafka

来源：https://stackoverflow.com/questions/73049038/creating-a-new-filtered-stream-in-ksqldb-with-where-clause-doesnt-work

2条答案

按热度按时间

a5g8bdjr1#

ksql> CREATE STREAM original(k INT KEY, v VARCHAR) WITH (kafka_topic='original', partitions=2, format='json');

ksql> CREATE STREAM foo_filter AS SELECT * FROM original WHERE v='foo';

ksql> INSERT INTO original (k, v) VALUES (1, 'foo');
ksql> INSERT INTO original (k, v) VALUES (2, 'bar');
ksql> INSERT INTO original (k, v) VALUES (3, 'foo');
ksql> INSERT INTO original (k, v) VALUES (4, 'bar');

ksql> select * from foo_filter;
+-------------------------------------------------------------------------------+-------------------------------------------------------------------------------+
|K                                                                              |V                                                                              |
+-------------------------------------------------------------------------------+-------------------------------------------------------------------------------+
|1                                                                              |foo                                                                            |
|3                                                                              |foo                                                                            |
Query Completed
Query terminated

赞(0）回复(0）举报 2022-09-21

zazmityj2#

问题最终被证明是Ksqldb无法更新/删除与查询相关的队列。

Kafka需要配置delete.topic.enable=true。

赞(0）回复(0）举报 2022-09-21