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

Question

我是Kafka的新手，我正在构建一个使用Twitter API作为数据源的入门项目。我已经创建了一个生产者，它可以查询Twitter API，并使用字符串序列化器将数据发送到我的kafka主题，以获得键和值。我的Kafka Stream应用程序读取这些数据并进行字数统计，但也按Tweet的日期分组。这一部分是通过一个名为wordCounts的KTable来完成的，以利用它的upsert功能。这个KTable的结构是：

键：{Word：exampleWord,Date：exampleDate},Value：numberOfOccurences

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

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

然而，当我试图将此发送到我的第二个Kafka主题时，我得到了下面的错误。

序列化程序（键：org.apache.kafka.common.serialization.StringSerializer/值：org.apache.kafka.common.serialization.StringSerializer）与实际的键或值类型（键类型：com.google.gson.jsonObject/值类型：com.google.gson.jsonObject）不兼容。更改StreamConfig中的默认SERDE，或者通过方法参数提供正确的Serdes。

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

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（）之前执行一个键更改操作，所以它将创建一个重分区主题，并且对于该主题，它将依赖于默认的键值serdes，您已经将其设置为String serde。

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