问题标签 [ksqldb]

1. 实现 KSQL UDF “NOW()”
2. 创建流 my_notification_stream 作为 select * from my_event_stream 其中发生时间 + 阈值 < NOW()
3. 收听 my_notification_stream 主题

我正在尝试使用 ksql 创建一个流。

我得到以下输出。我错过了什么吗？

我从一个主题 i ksql 创建了一个流。该流具有如下字段。我可以查询不同的字段，例如：select category from fake-data-119. 我想知道如何从地图字段中获取单个项目，例如：状态？

来自源的数据是：

我正在使用下面的语句来创建流：

我有如下流，我想从中创建另一个流。我正在尝试以下命令，但出现以下错误。我错过了什么吗？

我的 kafka 主题是以这种格式推送数据（来自collectd）：

它是数组、整数和浮点数的组合……整个东西都在一个 json 数组中。结果，我有一段时间使用ksql对这些数据做任何事情。

当我创建一个“默认”流时

我得到这个结果：

任何选择都将返回 ROWTIME 和 ROWKEY 的 8 位十六进制值。

我花了一些时间试图提取 json 字段无济于事。我担心的是：

这个主题可能不能在ksql中使用吗？是否有一种技术可以解开外部数组以获取内部有趣的位？

我想从 kafka-console-consumer 命令获取窗口开始时间戳（这里是 1530008520000）。

它适用于 KSQL：

但它不适用于 kafka-console-consumer？

如何从 kafka-console-consumer 打印窗口开始时间（这里是 1530008520000）？

谢谢！

我stream1在KSQL（5.0 Beta 版）中创建了一个支持主题topic1和avro架构。我能够阅读有关topic1使用的所有消息kafka-avro-console-consumer。

然后，我基于消息和支持主题的格式创建stream2了名为. 我能够阅读有关使用的所有消息KSQLstream1jsontopic2topic2kafka-console-consumer

我基于消息格式和名为stream3. 但是，我无法阅读有关使用.KSQLstream2jsontopic3topic3kafka-console-consumer

使用kafkacatI 在各个分区上获得偏移量，topic3但没有打印任何实际消息。

看起来消息在主题中，但两者都kafkacat无法kafka-console-consumer打印。

尝试使用--from-beginning但--offset earliest --partition 0没有运气。

这是 KSQL 语句

附加信息

如果ksql我运行SET 'auto.offset.reset'='earliest';并运行，select * from stream1 limit 5;或者select * from stream2 limit 5我看到打印的记录但select * from stream3 limit 5不返回任何记录。

如果我跑，describe extended stream3我会得到

消息总数：212

这恰好是我发送给 topic1 的消息数

我有一个在 ec2 实例上运行的 ksql 服务器，我想从我的本地机器连接到它。但是，它总是不断抛出以下错误：

我已修改我的安全组的设置以允许所需端口（服务器正在运行的端口）上的入站流量。我也可以从我的本地端口远程登录。这方面的任何线索都会非常有帮助！！谢谢！

我正在使用 kafka-connect 将行从 mysql 表流式传输到 kafka 主题。

当我在我的 ksql 客户端上运行它时：

我得到一个看起来像这样的主题列表：

我想知道 的含义是什么Registered，对于通过 kafka-connect 创建的所有主题似乎总是错误的。

我还没有在文档中找到任何东西。我还注意到，一旦我从这个主题创建了一个 STREAM，Registered就会变成true.

我正在使用 kafka-connect 将 MySQL 表中的行流式传输到 kafka 主题中。这很好用。

然后我创建一个表：

这也有效，我可以通过以下方式确认：

我明白了mytable。

问题是当我执行

然后我没有得到任何结果并且提示没有响应，我必须按下ctrl+c才能重新获得控制权。

可能是什么问题呢？