Word count program using Kafka Streams

word-count-kafka-streams.java

KStreamBuilder builder = new KStreamBuilder();
KStream<String, String> textLines = builder.stream(stringSerde, stringSerde,"TextLinesTopic");
Pattern pattern = Pattern.compile("\\W+", Pattern.UNICODE_CHARACTER_CLASS);
KStream<String, Long> wordCounts = textLines
                                      .flatMapValues(value-> Arrays.asList(pattern.split(value.toLowerCase())))
                                      .map((key, word) -> new KeyValue<>(word, word))
                                      .countByKey("Counts")
                                      .toStream();
wordCounts.to(stringSerde, longSerde, "WordsWithCountsTopic");
KafkaStreams streams = new KafkaStreams(builder, streamsConfiguration);
streams.start();