我用这段代码打印来自测试主题3的每个数字的频率。我运行consumer在test\u topic\u 3中获取csv数据。

KStream<String,String> inputTopic=builder.stream("test_topic_3");
        KTable<String, Long> filtered = inputTopic
                .selectKey((k, v) ->
                {
                    System.out.println("key:"+k+"|| value:"+v.split(",")[1]);
                    return v.split(",")[1];
                }
                )
                .groupByKey()
                .count();

        System.out.println("-------");
        filtered.toStream().foreach((k,v)-> System.out.println(k+"|"+v));
        System.out.println("-------");