问题标签 [ksqldb]

我打算使用 KSQL 进行流处理。我有以下要求。

• 设置查询后，我想从查询中获取该输出并将其作为 JSON 发送到 RabbitMQ。

• 我有多个用户将为查询创建标准，我将采用该标准并将其转换为 KSQL 并提交给 KSQL 引擎。我想即时执行此操作，这样每当他们有新的过滤器要求时，我就不必一直插话自己。我将创建 UI 工具来捕获他们的标准，我的程序会将其转换为 KSQL 查询。

我还想保存来自多个用户的这些查询，以便在系统重新启动时我可以再次提交这些查询

我怎样才能做到这一点？

我正在尝试加入以下流

下表：

使用

但USERNAME只包含null值。

我确信有 common user_ids 并且没有username在null表中。我在这里错过了什么吗？

我是ksql的新手。我只是在玩阅读 kafka 主题到流中，效果很好。

此外，尝试从 kafka 主题创建表并失败。意识到我需要在 kafka 主题中设置一个键，该键被视为 ksql 表中的主键。所以我尝试从流创建表，但也失败了。查询/脚本：

有人可以解释它是否可能吗？如果是，我哪里错了？谢谢。

我有两个 Kafka Streams，其中包含我使用 KSQL 创建String的格式的键和值。Avro

这是第一个：

第二个：

所需的输出应包括IDUSER、LASTNAME和。DEVICEUSERNAME

我想使用 Streams API 处理left join这些流（on IDUSER）并将输出写入 kafka 主题。

为此，我尝试了以下方法：

然而，

在我的 IDE 上引发错误：

当我尝试将 aString Serde用于键和值时，它可以工作，但数据不能从kafka-console-consumer. 我想要做的是生成 AVRO 格式的数据，以便能够使用kafka-avro-console-consumer.

我创建了一个重新加密的流

从此流中选择，我得到 > 100 个项目

然后我尝试创建一个表

当我跑步时

我得到的只有不到 10 个喜欢的物品；

我运行命令：

一切都好。我得到喜欢的数据

值字节：158

问题出在哪里？

我在使用 Confluent 4.1 的流和表中重新输入了数据

1）创建流

2）创建重新加密的流，因为这个脚本不起作用，但在此之前它的工作，为什么？

然后我创建下一个脚本

session_details_stream_rekeyed 的结果是好的：

3）为主题创建流；

4）创建表

5）然后我创建一个左连接

哪里有问题？

我在使用 Confluent 开源平台 4.1.0 版时遇到以下问题：

更改侦听器端口没有帮助。我们如何解决这个问题？

EDIT1：我正在使用 kafka 代理和 ksql-server

最初，“confluent status”显示 ksql-server 是 UP，但在上述超时后服务器宕机。

EDIT2：是的，我的 kafka 代理正在运行，这是我的 kafka server.properties：

和 ksql-server.properties：

EDIT3：我怀疑这与不正确的引导服务器 url 有关，但我还没有找到。

EDIT4：KSQL 服务器日志，根据要求。

我正在尝试在 Apache/KAFKA KSQL 中创建一个流主题包含（有点复杂的 JSON）

现在，对于像 event_type 和 created_at 这样简单的东西，创建流很容易。那会是这样

CREATE STREAM tstream (event_type varchar, created_at varchar) WITH (kafka_topic='usage_events', value_format='json');

但是现在我需要访问 used_service_units....并且我想在上面的 JSON 中提取“金额”

我该怎么做？

结果是

如果我改为创建这样的流

然后像这样在流上执行 SQL SELECT

这些替代方案都不起作用...

这个给了我

在 Ksql 中创建自定义 UDAF 时遇到问题。用例是在翻滚窗口中查找列的“第一个”和“最后一个”值。UDAF 中没有这样的内置功能（https://docs.confluent.io/current/ksql/docs/syntax-reference.html#aggregate-functions），所以我正在尝试创建自定义 UDAF。

我根据本文档https://www.confluent.io/blog/write-user-defined-function-udf-ksql/执行了以下步骤

一世。创建了 UDAF & AggregateFunctionFactory 并在 FunctionRegistry 中注册如下：

ii.构建 ksql-engine jar 并在以下路径 $CONFLUENT_HOME/share/java/ksql 的 confluent 包中替换它。

iii.重启ksql-server

但是，似乎该功能未注册。有什么建议么？

汇合版本：4.1.0

注意：我尝试创建简单的 UDF 。效果很好。问题在于 UDAF

我正在尝试在无头模式下测试 ksql。我在查询文件中添加了 Create Table 查询并启动了服务器，

查询文件有以下查询。

但是，我在日志中收到以下异常。

但是，当我在交互模式下运行相同的查询时，它执行得很好。注意：流已经存在于 ksql 中。我正在运行 4.1.0 CLI 版本。