用于将 JSON 数组转换为 json 对象的 kafka 流

Question

我在普通 java 中有一个代码，用于将 JSON 数组转换为 JSON 对象，我需要将这个普通 java 转换为 kafka 流......下面是我的

import java.io.*;
import java.util.*;
import java.lang.*;
import org.json.simple.JSONArray;
import org.json.simple.JSONObject;
import org.json.simple.parser.JSONParser;
import org.json.simple.parser.ParseException;
public class JsonParseTest {
    public static void main(String[] args) {
        try {
            JSONParser parser = new JSONParser();
            JSONArray jsonArray = (JSONArray) parser.parse(new FileReader("/root/jsonTestFile.json"));
             for (Object o : jsonArray) {
//to get the Json object
              JSONObject snap = (JSONObject) o;
              System.out.println(snap);
             }
}
 catch (FileNotFoundException ex) {
            ex.printStackTrace();
        } catch (IOException ex) {
            ex.printStackTrace();
        } catch (ParseException ex) {
            ex.printStackTrace();
        } catch (NullPointerException ex) {
            ex.printStackTrace();
        }
    }
}

如果有人帮助我只为逻辑部分编写代码，我可以继续，下面是我的逻辑部分，至少我需要帮助

public class JsonParseTest {
    public static void main(String[] args) {
        try {
            JSONParser parser = new JSONParser();
            JSONArray jsonArray = (JSONArray) parser.parse(new FileReader("/root/jsonTestFile.json"));
             for (Object o : jsonArray) {
//to get the Json object
              JSONObject snap = (JSONObject) o;
              System.out.println(snap);
             }
}

如何在 kafka 流中编写相同的代码？有人可以帮忙吗？

Answer 1

我会推荐两个选项：

• 您实现自己的序列化器/反序列化器 (Here is the javadoc)

• 您可以将流处理为 (String,String) 的流，并使用 flatMap 解析流的每个元素并将其转换为 (String,JSONObject) 的流：

  JSONParser parser = new JSONParser();
  stringStream.flatMap((k,v) -> {
      List<KeyValue<String,JSONOBject>> tmp = new ArrayList<KeyValue<String,JSONOBject>>();
      JSONArray jsonArray = (JSONArray) parser.parse(v);
      for (Object o : jsonArray) {
           JSONObject snap = (JSONObject) o;
           tmp.add(new KeyValue(k, snap));
      }     
      return tmp;
  });
  

这里我根本没有处理必须将 lambda 代码包装到 try/catch 中的异常。