问题标签 [ksqldb]

我创建了这个记录：

我现在想做这样的事情：

是否有可能在 Stream 创建中访问密钥，或者我是否必须在值中也包含密钥？

我一直在使用 KSQL，到目前为止它运行良好。但现在我想通过 Kafka Connect 将输出下沉到 BigQuery，并且需要附加一个 JSON 模式。我很难弄清楚如何做到这一点。这是我的查询：

这是写入输出主题 ( tweets-new) 的记录示例。

但是，为了让 Kafka Connect 将这些记录接收到 BigQuery，我需要附加一个架构，如下所示：

无论如何，我在文档中没有看到任何东西表明我可以如何解决这个问题（也许我找错了地方）。任何帮助将不胜感激！

我们正在尝试将 Kafka KSQL 迁移到我们的系统，并希望分享一些我们在此过程中无法解决的问题。我们的集群中有 3 个 Kafka 节点，每个服务器都有：

在每台服务器上，我们都有 zookeeper 来管理集群。所有操作系统限制都增加了，因此节点可以使用比它需要的更多的资源：

目前，从生产者到集群的流量很小（每秒约 10 条消息）。现在我们只有一个生产者，消息格式是：

Kafka 中的主题分为 6 个分区和 1 个复制：

现在，当然，节点没有得到充分利用，在 kafka 方面一切都很好）

我们希望在 Kafka 之上使用 KSQL，以便能够使用 SQL 过滤进入我们系统的数据。以下是 KSQL 服务器资源：

我们只有一张桌子：

这是创建表的命令：

在我们的应用程序中，我们需要通过 user_id 订阅表，如下所示：

对于生产 KSQL 服务器配置，我们使用来自 confluent 的官方推荐： https ://docs.confluent.io/current/ksql/docs/installation/server-config/config-reference.html#recommended-ksql-production-settings

KSQL 服务器的操作系统和软件限制也有所增加：

如果我们只使用一个订阅，我们不会遇到任何问题（在这种情况下一切都很好）。
但我们总共需要超过 200000 个订阅。因此，当我们尝试获得 100-200 个并行订阅时，我们的客户端会出现“读取超时”。在服务器中，我们没有看到任何可能影响 KSQL 的异常负载。
我们假设这个问题只与 KSQL 有关，因为当我们尝试使用另一个 KSQL 服务器（在不同的机器上）时，同时我们可以看到第二个服务器工作正常并且可以处理大约 1-20 个订阅。

我在与 KSQL 服务器连接的 Internet 上找不到任何基准，在文档中，我也找不到任何提及 KSQL 用例的内容，也许它的设计目的只是为少量连接提供大量数据，或者我们的系统配置错误，因此我们应该修复它以使用该软件实现我们的目标。
任何建议都会有所帮助。
提前致谢 ）

我想连接到我自己的 Kafka 实例以尝试 KSQL。我下载了最新版本的 Confluent 开源平台 - https://www.confluent.io/download/

我遵循https://docs.confluent.io/current/ksql/docs/installation/server-config/index.html并尝试从 Git Bash 在我的 Windows 10 机器上启动 KSQL 服务器：

但它失败并出现以下错误：

我做错了什么以及如何正确启动它？

我有一个 Kafka 主题，其中包含客户、购买、产品等数据。我想从这个单一主题创建不同的流。

我们通过 CDC 将数据从 DB2（表 1）发送到 Kafka 主题（主题 1）。我们需要在 DB2 数据和 Kafka 主题之间进行协调。我们有两个选择——

a) 将所有 kafka 主题数据下载到 DB2 中（作为 table-1-copy），然后进行左外连接（在 table-1 和 table-1-copy 之间）以查看不匹配的记录，创建增量并推送它回到卡夫卡。 问题：可伸缩性——我们的数据集大约有 10 亿条记录，我不确定 DB2 DBA 是否会让我们运行如此庞大的连接操作（可能很容易持续超过 15-20 分钟）。

b) 将 DB2 再次推回并行 kafka 主题（topic-1-copy），然后执行一些基于 kafka 流的解决方案，以在 kafka topic-1 和 topic-1-copy 之间进行左外连接。我仍然围绕着卡夫卡流和左外连接。我不确定（在 kafka 流中使用窗口系统）我是否能够将 topic-1 的整个内容与 topic-1-copy 进行比较。

更糟糕的是，kafka 中的 topic-1 是一个 compact topic，所以当我们将数据从 DB2 推送回 Kafka topic-1-copy 时，我们无法确定性地启动 kafka topic-compaction 循环以确保两个 topic- 1 和 topic-1-copy 在对它们运行任何类型的比较操作之前完全压缩。

c) 有没有其他我们可以考虑的框架选项？

理想的解决方案必须针对任何大小的数据进行扩展。

嗨，我有一个 Ksql 表（NGINX_TABLE）的左连接 Ksql 流（SEARCHREQUESTDTO）。使用以下 ksql 命令

结果流 NIGINX_SEARCH_QUERY 成功。我也可以在 Ksql 终端中使用 show topic 命令查看 NIGINX_SEARCH_QUERY 主题。

当我尝试将 kafka 消费者连接到此主题时，消费者无法获取任何数据。

但是显示数据的打印 NIGINX_SEARCH_QUERY 命令正在此主题中发布。

我使用模式注册表将 avro 消息写入 kafka 主题。然后根据主题创建流。使用当前模式创建的尖叫声。

然后我将新字段添加到架构中。模式寄存器已更新，没关系，但流仍保留第一个结构。

我可以使用新架构更新流吗？

再次删除和创建模式对我来说是有问题的，因为我有很多其他依赖于它的流\表。KSQL 不允许丢弃具有依赖关系的流。

我有一个 kafka 主题，主题中的每条消息都有纬度/经度和事件时间戳。创建了一个引用主题的流，并希望使用 geo_distance 计算 2 点之间的距离。例子

我想在上面的流上创建一个新流并用距离丰富它。

使用 KSQL 是否可以达到预期的效果？或者如何在处理新消息时参考以前的消息？

您能否让我知道如何处理 KSQL 中的 Null 值。我试图处理 4 种可能的方法，但没有得到解决。尝试用 4 种方法在 KSQL 中用不同的值替换 NULL 但给出了问题。