ksql rest api创建并在json中获取输出答案

【问题标题】：ksql rest api creation and getting output in jsonksql rest api创建并在json中获取输出
【发布时间】：2019-09-17 22:42:05
【问题描述】：

我无法从 katable 获取数据到 pojo 类输出

我试图调用 confluent kafka 的 rest 端点，但我无法将输出放在 pojo 类中。

    @PostMapping("/QueryForEquipment")
    @Consumes("application/json")
    public String getEquipments() {
            String walletBalanceUrl = "http://172.21.79.18:8088/query";

            HttpHeaders httpHeaders = new HttpHeaders();
            httpHeaders.set("Content-Type", "application/json");
            httpHeaders.add("Accept", MediaType.APPLICATION_JSON.toString());

            JSONObject json = new JSONObject();
            json.put("ksql", "Select * from  EQP_STREAM limit 10;");

            JSONObject jsonSub = new JSONObject();
            jsonSub.put("ksql.streams.auto.offset.reset", "earliest");
            json.put("streamsProperties", jsonSub);

         /*   String body=new String("{
                    "ksql": "Select ROWTIME,ROWKEY,TRN_ID_KEY from TABLE_EQP_LOCATION limit 10;",
                "streamsProperties": {"ksql.streams.auto.offset.reset": "earliest"} }"); 

         */

            System.out.println(walletBalanceUrl);

            HttpEntity httpEntity = new HttpEntity(body, httpHeaders);

            RestTemplate restTemplate = new RestTemplate();
            System.out.println("json text====" + body);

            ResponseEntity<String> uri = restTemplate.postForEntity(walletBalanceUrl, httpEntity, String.class);

            return uri.getBody().toString();

预期的输出必须是 json 格式目前它正在作为

{"row":{"columns":[1556458915675,"CDAU603054","Q1214320190418","TRAIN","Arrived","BELLEVILLE","Q","1214","4",1555545600000," ",1555618200000,"SRS_TRAIN","-77.374856","44.179657",null,null,"CDAU603054","CDAU","603054","INTERMODAL","L","KC7","7","DTTX720029","DTTX","720029","C","T","TOP",1555621693617]},"errorMessage":null,"finalMessage":null}
{"row":{"columns":[1556458915680,"CDAU603066","Q1214320190418","TRAIN","Arrived","BELLEVILLE","Q","1214","4",1555545600000," ",1555618200000,"SRS_TRAIN","-77.374856","44.179657",null,null,"CDAU603066","CDAU","603066","INTERMODAL","L","KC7","8","DTTX720030","DTTX","720030","D","T","TOP",1555621693617]},"errorMessage":null,"finalMessage":null}
{"row":{"columns":[1556458915680,"CDAU603070","Q1214320190418","TRAIN","Arrived","BELLEVILLE","Q","1214","4",1555545600000," ",1555618200000,"SRS_TRAIN","-77.374856","44.179657",null,null,"CDAU603070","CDAU","603070","INTERMODAL","L","KC7","8","DTTX720030","DTTX","720030","A","T","TOP",1555621693617]},"errorMessage":null,"finalMessage":null}

【问题讨论】：

您显示的输出是 JSON。您能否举例说明您希望看到的格式？

标签： ksqldb

【解决方案1】：

这里，Content-Type 将是 application/vnd.ksql.v1+json 并设置charset 到 utf-8。

    HttpHeaders headers = new HttpHeaders();
    headers.set("Content-Type", "application/vnd.ksql.v1+json");
    headers.set("charset", "utf-8");
    headers.setAccept(Collections.singletonList(MediaType.APPLICATION_JSON));

而 ksql 查询将是 SELECT * from EQP_STREAM EMIT CHANGES limit 10;

     json.put("ksql", "SELECT * from  EQP_STREAM EMIT CHANGES limit 10;")

参考链接：https://docs.ksqldb.io/en/latest/developer-guide/ksqldb-rest-api/ksql-endpoint/

【讨论】：