StreamProcessingTopology.java

KStream<String, String> source = builder.stream("streams-file-input");

KTable<String, Long> counts = source
       .flatMapValues(new ValueMapper<String, Iterable<String>>() {
           @Override
           public Iterable<String> apply(String value) {
               return Arrays.asList(value.toLowerCase(Locale.getDefault()).split(" "));
           }
       }).map(new KeyValueMapper<String, String, KeyValue<String, String>>() {
           @Override
           public KeyValue<String, String> apply(String key, String value) {
               return new KeyValue<>(value, value);
           }
       })
       .groupByKey()
       .count("Counts");

counts.to(Serdes.String(), Serdes.Long(), "streams-wordcount-output");