SpringCloud/spring-cloud-stream-samples
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

kafka streams word count application cleanup

Browse Source
This commit is contained in:
Soby Chacko
2018-03-21 20:19:52 -04:00
parent d1169eeef7
commit 477d712214
2 changed files with 4 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
kafka-streams-samples/kafka-streams-word-count/README.adoc
View File

@@ -6,6 +6,7 @@ The example is based on the word count application from the https://github.com/c
It uses a single input and a single output.
In essence, the application receives text messages from an input topic and computes word occurrence counts in a configurable time window and report that in an output topic.
This sample uses lambda expressions and thus requires Java 8+.
The sample uses a default timewindow of 30 seconds.
=== Running the app:
@@ -15,7 +16,7 @@ Go to the root of the repository.
`./mvnw clean package`
`java -jar target/kafka-streams-word-count-0.0.1-SNAPSHOT.jar --spring.cloud.stream.kafka.streams.timeWindow.length=60000`
`java -jar target/kafka-streams-word-count-0.0.1-SNAPSHOT.jar`
Assuming you are running the dockerized Kafka cluster as above.
@@ -29,13 +30,4 @@ On another terminal:
Enter some text in the console producer and watch the output in the console consumer.
Time window can be changed using the following property.
`spring.cloud.stream.kafka.streams.timeWindow.length` (value is expressed in milliseconds)
In order to switch to a hopping window, you can use the `spring.cloud.stream.kstream.timeWindow.advnceBy` (value in milliseconds).
This will create an overlapped hopping windows depending on the value you provide.
Here is an example with 2 overlapping windows (window length of 10 seconds and a hop (advance) by 5 seconds:
`java -jar target/kafka-streams-word-count-0.0.1-SNAPSHOT.jar --spring.cloud.stream.kafka.streams.timeWindow.length=10000 --spring.cloud.stream.kafka.streams.timeWindow.advnceBy=5000`
Once you are done, stop the Kafka cluster: `docker-compose down`

6
kafka-streams-samples/kafka-streams-word-count/src/main/java/kafka/streams/word/count/KafkaStreamsWordCountApplication.java
View File

@@ -22,7 +22,6 @@ import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.Materialized;
import org.apache.kafka.streams.kstream.Serialized;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.stream.annotation.EnableBinding;
@@ -43,9 +42,6 @@ public class KafkaStreamsWordCountApplication {
@EnableBinding(KafkaStreamsProcessor.class)
public static class WordCountProcessorApplication {
@Autowired
private TimeWindows timeWindows;
@StreamListener("input")
@SendTo("output")
public KStream<?, WordCount> process(KStream<Object, String> input) {
@@ -54,7 +50,7 @@ public class KafkaStreamsWordCountApplication {
.flatMapValues(value -> Arrays.asList(value.toLowerCase().split("\\W+")))
.map((key, value) -> new KeyValue<>(value, value))
.groupByKey(Serialized.with(Serdes.String(), Serdes.String()))
.windowedBy(timeWindows)
.windowedBy(TimeWindows.of(30000))
.count(Materialized.as("WordCounts-1"))
.toStream()
.map((key, value) -> new KeyValue<>(null, new WordCount(key.key(), value, new Date(key.window().start()), new Date(key.window().end()))));