我目前正在开发一个 POC 项目，该项目使用 Kafka Connect 从 Kafka 主题中读取数据以写入数据库。我知道 JDBC 接收器连接器需要架构才能工作。然而，我们所有的 kafka 主题都是字符串分隔的。除了使用 json 或 avro 创建一个新主题之外，我还计划创建一个将字符串转换为 json 的 API，我可以尝试任何可能的解决方案吗？
【问题讨论】：
标签： apache-kafka apache-kafka-connect
您的问题的解决方案是创建一个自定义转换器。不幸的是，没有博客文章或在线资源说明如何逐步完成此操作，但您可以查看StringConverter 是如何构建的。这是一个非常简单的 API，它使用序列化器和反序列化器来转换数据。
L.E：
转换器使用序列化器和反序列化器，因此您唯一应该做的就是创建自己的：
public class StringConverter implements Converter {
private final CustomStringSerializer serializer = new CustomStringSerializer();
private final CustomStringSerializer deserializer = new CustomStringSerializer();
--- boilerplate code resulted by implementing the converter Interface ---
@Override
public byte[] fromConnectData(String topic, Schema schema, Object value) {
try {
return serializer.serialize(topic, value == null ? null : value.toString());
} catch (SerializationException e) {
throw new DataException("Failed to serialize to a string: ", e);
}
}
@Override
public SchemaAndValue toConnectData(String topic, byte[] value) {
try {
return new SchemaAndValue(createYourOwnSchema(topic, value), deserializer.deserialize(topic, value));
} catch (SerializationException e) {
throw new DataException("Failed to deserialize string: ", e);
}
}
}
您会注意到toConnectData 正在返回一个SchemaAndValue 对象，实际上它正是您需要自定义的。您可以创建自己的 createYourOwnSchema 方法，该方法将根据主题（和值？）创建架构，然后您拥有反序列化器来反序列化您的数据。
【讨论】：
如果我了解您的用例，我肯定会尝试使用 Kafka Streams 来处理您的分隔字符串主题并将其放入 JSON 主题中。然后，您可以将此 JSON 主题用作数据库的输入。
使用 Kafka Streams，您可以映射主题的每个字符串记录并将逻辑应用到其中以生成包含数据的 JSON。处理后的记录可以像 JSON 格式的字符串或什至使用 Kafka JSON Serdes (Kafka Streams DatatTypes) 的 JSON 类型一样沉入您的结果主题。
这种方法还增加了输入数据的灵活性，您可以处理、转换或忽略字符串的每个分隔字段。
如果您想开始使用，请查看Kafka Demo 和Confluent Kafka-Streams examples，了解更复杂的用例。
【讨论】：
让我们创建 Kafka Streams 应用程序。您将使用数据作为
KStream<String, String>（假设密钥也是String）
而不是在 Kafka Stream 应用程序中将其转换为 KStream<String,YourAvroClass> 并生成消息以针对 Avro 主题。
KStream<String,YourAvroClass> avroKStream ....
avroKStream.to("avro-output-topic", Produced.with(Serdes.String(), yourAvroClassSerde));
【讨论】：