如何使用KSQL将KTable的所有行输出到Kafka Topic答案

【问题标题】：How to output all rows of KTable to Kafka Topic using KSQL如何使用KSQL将KTable的所有行输出到Kafka Topic
【发布时间】：2021-07-16 19:39:26
【问题描述】：

我是 Ksql 新手并尝试聚合，创建了一个 Kstream 和一个 Ktable，看起来像这样

kstream...

 CREATE STREAM test
  (id BIGINT,
   type VARCHAR,
   sales BIGINT
  WITH (KAFKA_TOPIC='test1',
        VALUE_FORMAT='JSON');

ktable...

CREATE TABLE test_total AS
SELECT ID,
       SUM(SALES) AS TOTAL_SALES
FROM test
GROUP BY ID
EMIT CHANGES;

将这些值发布到 Kafka test1 主题中

{"id": 1, "type": "test","sales": 200}
{"id": 1, "type": "test","sales": 300}

当我使用 Kafka 控制台使用者时，我只看到输出为

{"TOTAL_SALES":200} {"TOTAL_SALES":500}

我怎样才能看到 id 也打印到 Kafka 主题中？我必须从表格中创建某种视图吗？

【问题讨论】：

【解决方案1】：

在您的控制台消费者命令中添加

--property print.key=true

如果你已经在使用 ksql，你也可以从你创建的 TABLE 中选择

【讨论】：

SELECT 工作正常，最终目标是将 Ktable 的输出放入某个数据库中。所以我想看看它在 Kafka Consumer 中的样子。添加 --property print 会添加奇怪的字符“”
在 Ktable 定义中添加了 KEY_FORMAT='JSON'，现在我可以看到整数 1 或 2 但没有 ID。它显示如下 - 1 {"TOTAL_SALES":600}。我的 Kafka 连接也因这种格式而崩溃。看起来 KTable 定义中缺少某些内容，并且与 Kafka 消费者控制台打印无关？
您可以使用 SMT HoistField ，请发布新问题并附上示例和您遇到的错误
如果我的回答对你有用，你至少应该+1，谢谢
名声不好无法+1，只能接受回答

【解决方案2】：

在 ksqlDB 中，原始类型被序列化为主题中的原始类型。

atm 对此你无能为力。

【讨论】：