-----BEGIN PGP SIGNED MESSAGE-----
You just need to read you stream and apply an (windowed) aggregation
If you use non-windowed aggregation you will get "since the
beginning". If you use windowed aggregation you can specify the window
size as 1 hour and get those results.
One comment: it seems that you want to count *all* queries. To make
this work, you need to make sure all records are using the same key
(because Kafka Streams only supports aggregation over keyed streams).
Keep in mind, that this prohibits parallelization of you aggregation!
As a workaround, you could also do two consecutive aggregation, and do
parallelize the first one, and do not parallelize the second one (ie,
using the first one as a pre aggregation similar to a combine step)
Without pre aggregation and assuming all records use the same key
something like this (for current trunk):
> KStreamBuilder builder = new KStreamBuilder(): KStream input =
> KGroupedStream groupedInput = input.groupByKey();
groupedInput.count(TimeWindows.of(3600 * 1000),
For more details, please see the docs and examples:
On 10/18/16 5:00 AM, Furkan KAMACI wrote:
> I could successfully run Kafka at my environment. I want to monitor
> Queries per Second at my search application with Kafka. Whenever a
> search request is done I create a ProducerRecord which holds
> current nano time of the system.
> I know that I have to use a streaming API for calculation i.e.
> Kafka Streams or Spark Streams. My choice is to use Kafka Streams.
> For last 1 hours, or since the beginning, I have to calculate the
> queries per second. How can I make such an aggregation at Kafka
> Kind Regards, Furkan KAMACI
-----BEGIN PGP SIGNATURE-----
Comment: GPGTools - https://gpgtools.org
-----END PGP SIGNATURE-----