问题标签 [ksqldb]

我正在用一张桌子加入一个流。连接的结果只是部分成功。一些值被准确填充，而另一些为空。我检查以确保表和流中都存在这些值，并且用于连接的列是两者的键。

我正在使用 confluent 加载使用 jdbc 从 MSSQL 中的表中读取的主题。

然后我使用 KSQL 从相应的主题创建流和表，数据都是 JSON 格式。

表中缺失数据的时间戳早于流的时间戳。

有什么建议么？

溪流 ：

对于运行时统计信息和查询详细信息，请运行：DESCRIBE EXTENDED <Stream,Table>;

表 01：

尝试加入流与 table-01 按预期工作。创建流 finalstream 作为a.id从where isammas的左连接appat b中选择。a.id = b.idb.idnull

但是，当我尝试根据以下查询将多个表与流连接时：

出了什么问题？谢谢。

我使用创建了一个 ksql 流

我有一个生产者正在运行，将数据泵入流中。是否可以列出/显示发送到 ksql 流的最新项目？

谢谢

PS我在ksql server中试过，结果是

当我尝试

它永远挂在那里

我正在探索 confluent Kafka 并遵循 confluent 中的此快速入门指南。java.lang.ClassNotFoundException: io.confluent.monitoring.clients.interceptor.MonitoringProducerInterceptor每当我尝试使用以下命令为 Kafka 主题创建一些示例数据时，我都会得到：

这是堆栈跟踪

我已按照快速入门指南中给出的每个步骤进行操作，我做错了什么？

我什至重申了融合，但仍然得到与这个谷歌小组答案中提到的错误相同的错误

我有我在 mongo db 中加载的示例 Json 文件，以使用 debezium 连接器发送到 Kafka 连接。我的问题是，尽管原始 json 输入文件中有有效数据，但如何将 RowKey 和 id 键列值转换为 0（ Zeros ）？以下是产生此错误的步骤。

第 1 步：创建示例 json 文件

在customerprofile集合中将此数据加载到mongodb中

第2步：创建mongodb debezium连接器配置并在kafka connect中激活连接器

第三步：检查主题队列

正如您在下面的输出 debezium 中看到的那样，由带有 objectid 的 id 列转换

****我的问题是因为我们使用 JsonConvetor 作为密钥转换器，我的 id 列如何用序列号密钥更新？****

第 4 步：从主题队列创建一个流

第 5 步：这是流从 CUSTOMER_REPART 接收零（0）的地方

第 6 步：请检查此打印命令的输出，它在 ROWKEY 和 id 键列值中显示零。他们是从哪里来的？我需要做些什么来确保 ROWKEY 和 id 键列显示原始 json 文件值中的正确值，这些值最初加载到 mongodb 中？

"Male","COMMENTS":"反向客户端-服务器联盟","ID":0,"LAST_NAME":"Suche","EMAIL":"csuche5@cbsnews.c {"ROWTIME":1533154567104,"ROWKEY" :"0","CLUB_STATUS":"platinum","GENDER":"Male","COMMENTS":"分阶段统一软件","ID":0,"LAST_NAME":"Suttling","EMAIL":" msuttling7@百度。{"ROWTIME":1533154567104,"ROWKEY":"0","CLUB_STATUS":"gold","GENDER":"Female","COMMENTS":"注重质量的全球软件","ID":0," LAST_NAME":"Stepney","EMAIL":"estepney8@we 阶段统一软件","ID":0,"LAST_NAME":"Suttling","EMAIL":"msuttling7@baidu. {"ROWTIME":1533154567104,"ROWKEY":"0","CLUB_STATUS":"gold","GENDER":"Female","COMMENTS":"注重质量的全球软件","ID":0," LAST_NAME":"Stepney","EMAIL":"estepney8@we 阶段统一软件","ID":0,"LAST_NAME":"Suttling","EMAIL":"msuttling7@baidu. {"ROWTIME":1533154567104,"ROWKEY":"0","CLUB_STATUS":"gold","GENDER":"Female","COMMENTS":"注重质量的全球软件","ID":0," LAST_NAME":"Stepney","EMAIL":"estepney8@we

也许是一个幼稚的问题，但任何人都可以为我提供 KSQL 的 sbt 依赖项吗？

我检查了Maven，但找不到任何东西。

依赖项是否托管在 Maven 以外的其他地方，如果是，我必须在 build.sbt 文件中添加的左轮手枪是什么？

我正在尝试编写一个 Scala 应用程序，它使用 Ksql 来查询一些 kafka 主题，以创建一个带有一些指标的仪表板。

我对 Confluent KSql 很陌生，但对 Kafka 并不陌生。我有作为 Avro 序列化数据存在于 Kafka 中的现有主题。我已经启动并运行了 Confluent 模式注册表，并将 KSql 配置为指向注册表。

当我尝试根据我的一个主题创建一个表时，KSql 抱怨它找不到流。当我尝试在 KSql 中创建一个仅在 KSql 中流式传输我的主题的流时，似乎无法指向在注册表中有引用的我的 Avro 序列化主题。

有谁知道如何解决这两个问题？我想使用 KSql 的方式不适合它的功能吗？

更新

这里有更多细节

KSql 配置

注册表配置

尝试通过声明外部主题来解决问题

我有两个关于在使用窗口的 KSQL wrt 查询中查询的问题：

1. 假设我有以下聚合查询：

   SELECT id, COUNT(*) FROM testtopic_stream WINDOW TUMBLING (SIZE 30 DAYS) GROUP BY id;

上述聚合的结果是仅使用进来的新报价计算得出的，还是实际上会遍历过去 30 天的所有数据，然后执行聚合？

2. 查询的最大可能窗口大小是多少？我看到我可以设置一个窗口甚至 30 天，并且查询现在似乎工作正常。有推荐的最大窗口大小吗？

我正在使用如下所示的查询来跟踪特定字段的滚动总和：

现在，对于每个输入刻度，它似乎返回了 6 个不同的聚合值，我猜它们是针对以下时间段的：

如果我有兴趣只为每个进入的滴答获得 [start, start+60] 秒的结果。有没有办法只得到那个？

我有一个 json 结构化数据，例如

{"name": "foo", "hobbies": ["music", "food", "travel"]}

我的问题是如何将其转换为 KSQL 中的 3 条记录：

{"name":"foo", "hobby":"music"} {"name":"foo", "hobby":"food"} {"name":"foo", "hobby":"travel" }