如何将 SinkRecord 转换为 JSON 字符串？

Question

假设 myAPICreate 需要一个 JSON 字符串。

public void put(Collection<SinkRecord> collection) {
        for (SinkRecord record : collection) {
            JSONObject recordJson = toJSON(record.value());
            String recordJsonString = recordJson.toString();

            myAPICreate(recordJsonString);
        }
    }

toJSON 是我定义的一个助手，它只记录记录，returns 一个 JSONObject.

JSONObject json = new JSONObject()
  .put("a", record.getString("a"))
  .put("b", record.getString("b"))
  .put("c", record.getString("c"));

我觉得我可能在这里做了很多多余的工作。是否有必要将 put 中的代码转换为 JSON 或是否有使用转换器的方法，以便 record 已经以 JSON 或 JSON 字符串？那我就可以通过myAPICreate(record.value().toString())而不需要手动去做？

Answer 1

当您 create a SinkRecord, you have a key & value schema w/ a key and value Object. Those objects should be Struct instances 时必须创建具有匹配的 Schema

在连接器配置中，您将使用 JSONConverter（或其他转换器）来获取序列化输出

如何将 SinkRecord 转换为 JSON 字符串？

How to convert SinkRecord to JSON string?

apache-kafka

apache-kafka-connect