问题标签 [ksqldb]

我正在尝试使用 Kafka Streams 来实现用例。我在 MySQL 中有两个表 - 用户和帐户。我正在使用 Kafka MySQL 连接器将事件从 MySQL 获取到 Kafka。

我需要从 Kafka 内部获取帐户中的所有用户 ID。所以我打算在 MySQL 输出主题上使用KStream，处理它以形成一个输出并将其发布到一个主题，其中 Key 为 account-id，value 为 userId，用逗号（，）分隔。然后我可以通过ReadOnlyKeyValueStore类的get()方法使用交互式查询使用帐户 ID 获取所有用户 ID 。这是正确的方法吗？有没有更好的办法？这里可以使用 KSQL 吗？

是否可以使用 KSQL 不仅通过计算特定列GROUP BY的条目，还可以获取通过应用程序流式传输的所有条目的聚合？

我正在寻找这样的东西：

或者在 KSQL 中更像这样：

谢谢
- 蒂姆

脚步 ：

1. 我们创建了一个名为 Kafka 的主题pgsqlcountry，其中包含来自 postgreSQL DB 的所有流数据。

2. 我们创建了一个流country，用于将主题处理成表格。

3. 流创建成功。

-

-

4. 我们运行 SQL 命令“select * from country”

5. 我们得到如下错误

-

请在此屏幕截图中找到我的堆栈跟踪

我对 KSQL 很陌生，只是想知道如何提取今天的数据进行聚合？假设我有以下模式的计数数据流：

如何使用 KSQL 创建一个表，该表将输出今天 UNIX 时间戳COUNT在哪里的总和的快照？TS

有没有办法从 KSQL 中删除主题？根据github这是可能的，我试过了

但是这两个命令都不起作用。我收到消息

有没有办法查询 KTABLE 中的所有当前条目？我正在尝试使用有效负载执行对 REST api 的 http 请求

并且流无限期地挂起。文件说

它相当于传统的数据库表，但通过诸如窗口之类的流语义来丰富。

那么当您只需要所有当前数据而不需要流式传输并将 KTABLE 视为常规缓存表时，是否可以进行常规查询？

我有一个名为A的 Kafka 主题。

主题A中的数据格式为：

现在在消费者方面，我只想获取一小时窗口的最新数据意味着每隔一小时我需要从基于 created_at 的主题中获取最新值

我的预期输出是：

我认为这可以通过 ksql 解决，但我不确定。请帮我。

预先感谢。

也许这是一个初学者的问题，但是推荐的读取 KSQL 生成的数据的方法是什么？

假设我进行了一些流处理并将数据写入 KSQL 表。现在我想通过 Spring 应用程序访问这些数据（例如，通过 websocket 扇出一些实时数据）。我的第一个猜测是使用 Spring Kafka 并订阅底层主题。还是我应该使用 Kafka Streams？

另一个用例可能是进行流处理并将结果写入 Redis 存储（例如，对于始终返回当前值的 Web 服务）。这里的方法是什么？

谢谢！

我们正在使用 Docker 设置 ksql。但它抛出

这是 /etc/ksql/ksqlserver.properties 文件的内容：

“0e234781853f:9092,5e9fbcd03294:9092,c48090cb8e79:9092”是 kafka 代理容器。我们可以从 ksql 容器向代理生成消息。

我们使用“confluentinc/ksql-cli:0.3” dockerhub 镜像作为基础。那么，我的设置有什么问题吗？

在 nodered 中，我已经能够json stream使用KSQL node. 现在我正在尝试以 json 文件的形式将该流推送到 S3 存储桶，kafka-s3-connector但我只能使用 cli 来执行此操作，而不是使用SQL and S3 node安装在 Node-Red 中。其中是否缺少其他节点，请帮助相同的？