问题标签 [ksqldb]

我们的 JavaScript 程序每秒使用 kafka-rest 异步写入 20 条消息。我们尝试对传入消息进行聚合，但它返回一些不一致的结果。

请在下面找到主题、流和聚合结果表定义。

话题：

溪流：

NEW STREAM - 此流使用实际事件日期而不是在 kafka 中写入消息的时间。

现在我们使用以下方法根据状态聚合数据：

现在当我们运行查询 SELECT * FROM ORDERS_AGG_SEC; 它返回以下结果

预期结果：对于处理状态，我应该每 1 秒获得 20 个计数

实际结果：对于相同的状态，我每 1 秒间隔获得多个记录，如下所示：

请在下面找到我的 javascript 代码：

注意：kafka rest api 使用默认属性运行

我正在通过以下方式使用融合版本 3.3.0 设置 ksql-cli

我正在以下列方式运行 ksql-cli docker 映像

当我ksql-cli通过以下方式进入容器的 bash运行时

并ksql-cli以下列方式运行：

它给了我以下错误：

正在初始化 KSQL...

无法获取代理信息。KSQL 无法初始化 AdminCLient。

我的一个主题有一个 string-json 作为键 - {"city":"X","id":22}。在我的 ksql 语句中，我想将其提取到 2 个不同的字段而不是一个，以便以后过滤和加入。在文档中，它似乎允许我只将整个字符串粘贴到键中，而不是允许将其格式化为 JSON（就像 FORMAT_VALUE 一样），见下文......对于这个优雅的解决方案有什么建议吗？

VALUE_FORMAT（必需）指定主题中消息值的序列化格式。支持的格式：JSON、DELIMITED 和 AVRO

KEY 将 Kafka 主题中的消息键与 KSQL 流中的列相关联。

我已经检查了 2 个类似的问题，但没有帮助。

我的属性：

尝试添加以下内容，但仍然出现相同的错误。

我的主题是从 KSQL 流中填充的。

当我下载最新的融合使用时，我发现我可以运行......

环境是：

有什么问题？

我正在使用融合 4.0.1

我正在使用以下命令启动 KSQL CLI。

当我尝试查询流或创建新流时，出现以下错误。

我已经验证了所有配置，看起来不错。

我们在 0.5 版本中从来没有遇到过这样的错误。

我最近将 Confluent 升级到 4.1，但架构注册表似乎有一些问题。在confluent startschema-registry（因此 ksql-server）上无法启动。

这是我在模式注册表的日志中得到的错误：

我不知道为什么会报告此错误，并且错误消息对我来说没有意义。

失败后，confluent start schema-registry启动confluent start ksql-server两个服务，但是在启动 KSQL 时，我收到以下警告：

尝试运行命令（例如show tables;）时，会报告以下错误：

编辑：我已经通过破坏当前运行来解决这个问题（confluent destroy)但如果有人能解释这个问题会很有趣。

我在 KSQL 中有一个蒸汽和一个表，如下所述：

流名称：DEAL_STREAM

表名：EXPENSE_TABLE

当我运行以下查询时，它仅显示流中的列，但不显示表列。

这是预期的输出。如果不是我做错了什么？

查询的输出如下所示。

我正在尝试在 Kafka 主题之上在 KSQL 中创建一个 Kafka 流。我将如下 JSON 记录存储在 Kafka 主题中。

{ "venue": { "venue_name": "HATCH", "lon": -71.18291, "lat": 42.36667, "venue_id": 22491322 }, "visibility": "public", "response": "yes", "guests": 0, "member": { "member_id": 237655942, "member_name": "Nts" }, "rsvp_id": 1724941595, "mtime": 1524620970613, "event": { "event_name": "Intro to Soldering", "event_id": "250106100", "time": 1526853600000, "event_url": "https:\/\/www.meetup.com\/Makers-of-HATCH-Makerspace\/events\/250106100\/" }, "group": { "group_topics": [ { "urlkey": "quilting", "topic_name": "Quilting" }, { "urlkey": "robotics", "topic_name": "Robotics" }, { "urlkey": "sewing", "topic_name": "Sewing" }, { "urlkey": "edtech", "topic_name": "Education & Technology" }, { "urlkey": "craftswap", "topic_name": "Crafts" }, { "urlkey": "diy", "topic_name": "DIY (Do It Yourself)" }, { "urlkey": "hacking", "topic_name": "Hacking" }, { "urlkey": "3d-modeling", "topic_name": "3D Modeling" }, { "urlkey": "tools", "topic_name": "Tools" }, { "urlkey": "arduino", "topic_name": "Arduino" }, { "urlkey": "makers", "topic_name": "Makers" }, { "urlkey": "makerspaces", "topic_name": "Makerspaces" }, { "urlkey": "3d-printing", "topic_name": "3D Printing" }, { "urlkey": "laser-cutting", "topic_name": "Laser Cutting" }, { "urlkey": "scrapbook-die-cutting-machines", "topic_name": "Scrapbook die cutting machines." } ], "group_city": "Watertown", "group_country": "us", "group_id": 18457932, "group_name": "Makers of HATCH Makerspace", "group_lon": -71.18, "group_urlname": "Makers-of-HATCH-Makerspace", "group_state": "MA", "group_lat": 42.37 } }

此数据已加载到 Kafka 主题中。

我在 KSQL 中创建了一个流，如下所示。

我在 group_info（kafka 流中的最后一个字段）字段中看到 null。注意：卡夫卡不允许我创建一个名为“组”的字段，因为它是一个关键字。因此将该字段命名为 group_info。

不知道我做错了什么，但欢迎提出任何建议。

第 1 步：运行生产者以创建示例数据

样本数据 ：

步骤 2：打开另一个终端并运行消费者来测试数据。

步骤 3：打开另一个终端并运行生产者。

数据：

第 4 步：打开另一个终端并运行消费者

第 5 步：登录到 KSQL 客户端。

创建以下流和表并运行连接查询。

KSQL：

溪流：

桌子：

以下是输出：