a2fea3a188
Add sample for the canonical WordCount Showing how to track group of products Showing how to implement interactive queries Fix #41 Update scst versions Enhance kstream samples
59 lines
2.9 KiB
Plaintext
59 lines
2.9 KiB
Plaintext
== What is this app?
|
|
|
|
This is an example of a Spring Cloud Stream processor using Kafka Streams support.
|
|
|
|
The example is based on a contrived use case of tracking products.
|
|
Although contrived, this type of use cases are pretty common in the industry where some organizations want to track the statistics of some entities over a time window.
|
|
In essence, the application receives product information from input topic and count the interested products in a configurable time window and report that in an output topic.
|
|
This sample uses lambda expressions and thus requires Java 8+.
|
|
|
|
==== Starting Kafka in a docker container
|
|
|
|
* Skip steps 1-3 if you already have a non-Docker Kafka environment.
|
|
|
|
1. Go to the docker directory in this repo and invoke the command `docker-compose up -d`.
|
|
2. Ensure that in the docker directory and then invoke the script `start-kafka-shell.sh`
|
|
3. cd $KAFKA_HOME
|
|
4. Start the console producer: +
|
|
Assuming that you are running kafka on a docker container on mac osx. Change the zookeeper IP address accordingly otherwise. +
|
|
`bin/kafka-console-producer.sh --broker-list 192.168.99.100:9092 --topic products`
|
|
5. Start the console consumer: +
|
|
Assuming that you are running kafka on a docker container on mac osx. Change the zookeeper IP address accordingly otherwise. +
|
|
`bin/kafka-console-consumer.sh --bootstrap-server 192.168.99.100:9092 --key-deserializer org.apache.kafka.common.serialization.IntegerDeserializer --property print.key=true --topic product-counts`
|
|
|
|
=== Running the app:
|
|
|
|
Go to the root of the repository and do:
|
|
|
|
`./mvnw clean package`
|
|
|
|
`java -jar target/kstream-product-tracker-0.0.1-SNAPSHOT.jar --kstream.product.tracker.productIds=123,124,125 --spring.cloud.stream.kstream.timeWindow.length=60000 --spring.cloud.stream.kstream.timeWindow.advanceBy=30000 --spring.cloud.stream.bindings.input.destination=products`
|
|
|
|
The above command will track products with ID's 123,124 and 125 every 30 seconds with the counts from the last minute.
|
|
In other words, every 30 seconds a new 1 minute window is started.
|
|
|
|
|
|
* By default we use the docker container IP (mac osx specific) in the `application.yml` for Kafka broker and zookeeper.
|
|
Change it in `application.yml` (which requires a rebuild) or pass them as runtime arguments as below.
|
|
|
|
`spring.cloud.stream.kstream.binder.brokers=<Broker IP Address>` +
|
|
`spring.cloud.stream.kstream.binder.zkNodes=<Zookeeper IP Address>`
|
|
|
|
Enter the following in the console producer (one line at a time) and watch the output on the console consumer:
|
|
|
|
```
|
|
{"id":"123"}
|
|
{"id":"124"}
|
|
{"id":"125"}
|
|
{"id":"123"}
|
|
{"id":"123"}
|
|
{"id":"123"}
|
|
```
|
|
|
|
The default time window is configured for 30 seconds and you can change that using the following property.
|
|
|
|
`kstream.word.count.windowLength` (value is expressed in milliseconds)
|
|
|
|
In order to switch to a hopping window, you can use the `kstream.word.count.advanceBy` (value in milliseconds).
|
|
This will create an overlapped hopping windows depending on the value you provide.
|