如何让KSQLDB Rest API返回JSON格式的查询结果？

当然可以实现！你只需要在KSQL查询语句中添加FORMAT JSON选项，就能让返回的每行数据变成你想要的JSON对象格式，完美适配POCO反序列化需求。

修改后的请求命令 #

直接修改你的curl请求中的SQL语句，加上FORMAT JSON：

curl --http2 -X "POST" "http://localhost:8088/query-stream" -H "Content-Type: application/vnd.ksql.v1+json; charset=utf-8" -d $'{ 
  "sql": "SELECT * FROM USERS EMIT CHANGES FORMAT JSON;", 
  "streamsProperties": { 
    "ksql.streams.auto.offset.reset": "earliest" 
  } 
}'

返回结果示例 #

执行后你会得到这样的输出：

{"queryId":"cdfb3ccc-0ab5-4186-a249-b279bfc09587","columnNames":["USERID","NAME"],"columnTypes":["STRING","STRING"]}
{"USERID":"1","NAME":"Max"}
{"USERID":"2","NAME":"Alex"}
{"USERID":"13","NAME":"Andrew"}
...

自定义小写键名（可选） #

如果需要像示例中那样用小写的userid、name作为键名，只需要在查询中给列设置别名即可：

SELECT USERID AS userid, NAME AS name FROM USERS EMIT CHANGES FORMAT JSON;

对应的curl请求就是：

curl --http2 -X "POST" "http://localhost:8088/query-stream" -H "Content-Type: application/vnd.ksql.v1+json; charset=utf-8" -d $'{ 
  "sql": "SELECT USERID AS userid, NAME AS name FROM USERS EMIT CHANGES FORMAT JSON;", 
  "streamsProperties": { 
    "ksql.streams.auto.offset.reset": "earliest" 
  } 
}'

此时返回的结果就是你想要的格式：

{"queryId":"cdfb3ccc-0ab5-4186-a249-b279bfc09587","columnNames":["userid","name"],"columnTypes":["STRING","STRING"]}
{"userid":"1","name":"Max"}
{"userid":"2","name":"Alex"}
{"userid":"13","name":"Andrew"}
...

旧版本KSQL兼容方案（如果适用） #

如果你的KSQL版本低于5.3（不支持FORMAT JSON语法），可以先创建一个指定JSON输出格式的流，再查询这个流：

CREATE STREAM USERS_JSON WITH (VALUE_FORMAT='JSON') AS SELECT * FROM USERS;

然后查询这个新流：

curl --http2 -X "POST" "http://localhost:8088/query-stream" -H "Content-Type: application/vnd.ksql.v1+json; charset=utf-8" -d $'{ 
  "sql": "SELECT * FROM USERS_JSON EMIT CHANGES;", 
  "streamsProperties": { 
    "ksql.streams.auto.offset.reset": "earliest" 
  } 
}'

这种方式同样能得到JSON对象格式的输出，只是多了一步预定义流的操作。

内容的提问来源于stack exchange，提问作者Markus