# Average for each window

**URL:** <https://forum.confluent.io/t/average-for-each-window/5203>\
**Category:** Kafka Streams\
**Created:** [3 June 2022 13:39 UTC](https://forum.confluent.io/t/average-for-each-window/5203 "2022-06-03T13:39:56Z")\
**Posts on this page:** 4\
**Page:** 1

<div class="post-metadata">

**Author:** ![miele975](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/miele975/32/1948_2.png) [@miele975](https://forum.confluent.io/u/miele975)\
**Post date:** [3 June 2022 13:39 UTC](https://forum.confluent.io/t/average-for-each-window/5203/1 "2022-06-03T13:39:56Z")

</div>

Hi all,  
I’m new in kafka streams and I’m really going crazy. I have a stream of counter values represented by \<counter-name, counter-value, timestamp\>. I want to calculate the average value for each day, like this:

counterValues topic content:

```auto
"cpu", 10, "2022-06-03 17:00"
"cpu", 20, "2022-06-03 18:00"
"cpu", 30, "2022-06-04 10:00"
"memory", 40, "2022-06-04 10:00"

```

and i want to obtain this output:

```auto
"cpu", "2022-06-03", 15
"cpu", "2022-06-04", 30
"memory", "2022-06-04", 40

```

This is a snippet of my code that it doesn’t work (it seems to calculate count)…

```auto
		counterValueStream
			.groupByKey()
			.windowedBy(tumblingWindow)
			.aggregate(StatisticValue::new, (k, counterValue, statisticValue) -> {
	            statisticValue.setSamplesNumber(statisticValue.getSamplesNumber() + 1);
	            statisticValue.setSum(statisticValue.getSum() + counterValue.getValue());
	            return statisticValue;
	        }, Materialized.with(Serdes.String(), statisticValueSerde))
			.toStream()
	        .map((Windowed<String> key, StatisticValue sv) -> {
	            double avgNoFormat = sv.getSum() / (double) sv.getSamplesNumber();
	            double formattedAvg = Double.parseDouble(String.format("%.2f", avgNoFormat));
	            return new KeyValue<>(key.key(), formattedAvg) ;
	        })
	        .to("average", Produced.with(Serdes.String(), Serdes.Double()));

```

Can anyone help me, please?  
Thanks in advance!

---

<div class="post-metadata">

**Author:** ![mjsax](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/mjsax/32/3113_2.png) [@mjsax](https://forum.confluent.io/u/mjsax)\
**Post date:** [3 June 2022 18:30 UTC](https://forum.confluent.io/t/average-for-each-window/5203/2 "2022-06-03T18:30:53Z")

</div>

Please check out [Compute an average aggregation using Kafka Streams](https://developer.confluent.io/tutorials/aggregating-average/kstreams.html)

---

<div class="post-metadata">

**Author:** ![miele975](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/miele975/32/1948_2.png) [@miele975](https://forum.confluent.io/u/miele975)\
**Post date:** [5 June 2022 16:03 UTC](https://forum.confluent.io/t/average-for-each-window/5203/3 "2022-06-05T16:03:03Z")

</div>

Hi, I wrote the code starting from the project that you have mentioned…  
The window is the following:

```auto
		Duration windowSize = Duration.ofDays(1);
		TimeWindows tumblingWindow = TimeWindows.of(windowSize);

```

The content of counterValue topic and the ouput are:

 ![average](https://us1.discourse-cdn.com/flex019/uploads/confluentcommunity/original/2X/2/2185b9bcb834af9ed390f5863c170259fa98b13b.jpeg)

Note that I use a TimestampExtractor that use counter timestamp instead of kafka record.

I would like to have the following result:

```auto
“cpu”, “2022-06-03”, 15
“cpu”, “2022-06-04”, 30
“memory”, “2022-06-04”, 40

```

Maybe there is something I am not understanding about how windows work…

Thank you in advance!  
Elena

---

<div class="post-metadata">

**Author:** ![bbejeck](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/bbejeck/32/412_2.png) [@bbejeck](https://forum.confluent.io/u/bbejeck)\
**Post date:** [7 June 2022 18:03 UTC](https://forum.confluent.io/t/average-for-each-window/5203/4 "2022-06-07T18:03:20Z")

</div>

Hi @miele975

There are a couple of things to consider here.

If you’re not getting the results you’d expect, I’d first look into your aggregation code as Kafka Streams only executes the `Aggregator.apply` method.

One suggestion is to try using the [Topology Test Driver](https://docs.confluent.io/platform/current/streams/developer-guide/test-streams.html#testing-a-streams-application) on your streams application. Using the TTD you have complete control over the timestamps for your input records and that makes it easy to spot where the application is going sideways. The [aggregation tutorial](https://developer.confluent.io/tutorials/aggregating-average/kstreams.html?_ga=2.170470952.98262800.1654523091-1159438105.1626889744&_gac=1.57065176.1654268674.Cj0KCQjw4uaUBhC8ARIsANUuDjWd_ihXxCBJzzRXutGHjoyc7043MrjSCahX_7j99vSLBjtA7UWL1sUaAoVaEALw_wcB#write-a-test) you looked at contains a test as well you can use for guidance.

For the output, it seems like you want a [final result](https://www.confluent.io/blog/kafka-streams-take-on-watermarks-and-triggers/) for the day, i.e. when the window closes. Here’s an [example of suppression](https://github.com/confluentinc/learn-kafka-courses/blob/3978f93c342b0210b51941ab022f97811dd1ec21/kafka-streams/src/main/java/io/confluent/developer/windows/solution/StreamsWindows.java#L45-L54) from the [Kafka Streams course](https://developer.confluent.io/learn-kafka/kafka-streams/get-started/) on [Confluent Developer](https://developer.confluent.io/). Keep in mind with `suppress` that Kafka Streams emits results when there are incoming records that drive the `streamtime` forward, in other words the final window results aren’t automatically emitted when the window closes.

Let me know if you have any questions.

HTH,  
Bill