如何使Serdes与多步卡夫卡流一起工作

Question

我是Kafka的新手，我正在使用Twitterneneneba API作为数据源构建一个启动项目。我已经创建了一个Producer，它可以查询Twitterneneneba API，并使用键和值的字符串序列化器将数据发送到我的kafka主题。我的Kafka Stream应用程序读取这些数据并进行字数统计，但也按推特的日期进行分组。这部分是通过一个名为wordCounts的KTable来完成的，以利用其追加销售功能。这个KTable的结构是：

关键字：｛word:exampleWord，date:exampleDate｝，值：发生次数

然后，我尝试将KTable流中的数据重组为平面结构，以便稍后将其发送到数据库。您可以在wordCountsStructured KStream对象中看到这一点。这将重新构造数据，使其看起来像下面的结构。该值最初是JsonObject，但我将其转换为字符串以匹配我设置的Serdes。

Key: null, Value: {word: exampleWord, date: exampleDate, Counts: numberOfOccurences}

然而，当我试图将此发送到我的第二个卡夫卡主题时，我收到了以下错误。

一个序列化程序(密钥：org.apache.kafka.commun.serialization.StringSerializer/value:org.apache.kafka.common.serialization.StringSerializer(不是与实际键或值类型兼容(键类型：com.google.gson.JsonObject/value-type:com.goog.gson.JsonObject(。更改StreamConfig中的默认序列号或提供正确的序列号通过方法参数。

我对此感到困惑，因为我发送给主题的KStream的类型是<String, String>。有人知道我该怎么解决吗？

public class TwitterWordCounter {
private final JsonParser jsonParser = new JsonParser();
public Topology createTopology(){
StreamsBuilder builder = new StreamsBuilder();

KStream<String, String> textLines = builder.stream("test-topic2");
KTable<JsonObject, Long> wordCounts = textLines
//parse each tweet as a tweet object
.mapValues(tweetString -> new Gson().fromJson(jsonParser.parse(tweetString).getAsJsonObject(), Tweet.class))
//map each tweet object to a list of json objects, each of which containing a word from the tweet and the date of the tweet
.flatMapValues(TwitterWordCounter::tweetWordDateMapper)
//update the key so it matches the word-date combination so we can do a groupBy and count instances
.selectKey((key, wordDate) -> wordDate)
.groupByKey()
.count(Materialized.as("Counts"));
/*
In order to structure the data so that it can be ingested into SQL, the value of each item in the stream must be straightforward: property, value
so we have to:
1. take the columns which include the dimensional data and put this into the value of the stream.
2. lable the count with 'count' as the column name
*/
KStream<String, String> wordCountsStructured = wordCounts.toStream()
.map((key, value) -> new KeyValue<>(null, MapValuesToIncludeColumnData(key, value).toString()));
KStream<String, String> wordCountsPeek = wordCountsStructured.peek(
(key, value) -> System.out.println("key: " + key + "value:" + value)
);
wordCountsStructured.to("test-output2", Produced.with(Serdes.String(), Serdes.String()));
return builder.build();
}
public static void main(String[] args) {
Properties config = new Properties();
config.put(StreamsConfig.APPLICATION_ID_CONFIG, "wordcount-application1111");
config.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "myIPAddress");
config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
config.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
config.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
TwitterWordCounter wordCountApp = new TwitterWordCounter();
KafkaStreams streams = new KafkaStreams(wordCountApp.createTopology(), config);
streams.start();
// shutdown hook to correctly close the streams application
Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
}
//this method is used for taking a tweet and transforming it to a representation of the words in it plus the date
public static List<JsonObject> tweetWordDateMapper(Tweet tweet) {
try{
List<String> words = Arrays.asList(tweet.tweetText.split("\W+"));
List<JsonObject> tweetsJson = new ArrayList<JsonObject>();
for(String word: words) {
JsonObject tweetJson = new JsonObject();
tweetJson.add("date", new JsonPrimitive(tweet.formattedDate().toString()));
tweetJson.add("word", new JsonPrimitive(word));
tweetsJson.add(tweetJson);
}
return tweetsJson;
}
catch (Exception e) {
System.out.println(e);
System.out.println(tweet.serialize().toString());
return new ArrayList<JsonObject>();
}
}
public JsonObject MapValuesToIncludeColumnData(JsonObject key, Long countOfWord) {
key.addProperty("count", countOfWord); //new JsonPrimitive(count));
return key;
}

Answer 1

因为您在groupBy((之前执行密钥更改操作，它将创建一个重新分区主题，对于该主题，它将依赖于默认密钥value serdes，您已将其设置为String Serde。

您可以将groupBy()调用修改为groupBy(Grouped.with(StringSerde,JsonSerde)，这将有所帮助。