# Running multiple aggregations against a topic **URL:** \ **Category:** Kafka Streams\ **Created:** [30 April 2021 21:36 UTC](https://forum.confluent.io/t/running-multiple-aggregations-against-a-topic/1639 "2021-04-30T21:36:27Z")\ **Posts on this page:** 4\ **Page:** 1
**Author:** ![send2r](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/send2r/32/880_2.png) [@send2r](https://forum.confluent.io/u/send2r)\ **Post date:** [30 April 2021 21:36 UTC](https://forum.confluent.io/t/running-multiple-aggregations-against-a-topic/1639/1 "2021-04-30T21:36:27Z")
I am trying to build an application that will satisfy following need. Let’s assume that I am getting certain price values for a given item(key) on an input topic. I have two different conditions to check - if the average price value for and item is less than x for n seconds I want to generate a low price alert for the item. If average price value of an item is above y for m seconds I want to generate a high price alert for the item. Can I define this using single streams application? Or the only way is to define multiple typologies or applications? Also, I want to interrupt the aggregation for a given item based on some external event - how to indicate to streams that the processing of a given item is no longer needed based on the event? Thanks in advance. ---
**Author:** ![Bruno](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/bruno/32/138_2.png) [@Bruno](https://forum.confluent.io/u/Bruno)\ **Post date:** [5 May 2021 07:33 UTC](https://forum.confluent.io/t/running-multiple-aggregations-against-a-topic/1639/2 "2021-05-05T07:33:22Z")
Hi @send2r and welcome! Two options come to my mind: 1. You can split your stream into multiple branches, for example: ```auto final KStream stream = builder.stream(INPUT_TOPIC, Consumed.with(Serdes.String(), Serdes.String())); stream.print(Printed.toSysOut().withLabel("Branch 1:")); stream.print(Printed.toSysOut().withLabel("Branch 2:")); ``` This code will print each record twice, once prefixed with `Branch 1:` and once prefixed with `Branch 2:`. Instead of the `print` operation you could have your windowed aggregations and a filter after the aggregations could decide if an alert is issued or not. 1. You could use a `KStream#transform()` or `KStream#transformValues()` which allows you to maintain a state store and to issue records when your conditions are satisfied. Regarding the interruption, if the external events can be read from a Kafka topic you could read these events with `KStream#transform()` or `KStream#transformValues()` (maybe `KStream#process() ` would also work) and set a flag in a state store that can be read from another `KStream#transformValues()` operation that also accesses the same state store. You could also consider enriching your aggregated event with the flag to avoid the state store. Best, Bruno ---
**Author:** ![send2r](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/send2r/32/880_2.png) [@send2r](https://forum.confluent.io/u/send2r)\ **Post date:** [11 May 2021 03:28 UTC](https://forum.confluent.io/t/running-multiple-aggregations-against-a-topic/1639/3 "2021-05-11T03:28:58Z")
> [@Bruno](#): > > Regarding the interruption, if the external events can be read from a Kafka topic you could read these events with `KStream#transform()` or `KStream#transformValues()` (maybe `KStream#process() ` would also work) and set a flag in a state store that can be read from another `KStream#transformValues()` operation that also accesses the same state store. Thank you very much for the suggestions. I did not quite follow the last part regarding suspending/stopping streams processing based on an external event though. Is there an example somewhere that you could point me to? Also is it safe to assume that I can scale this topology by increasing partitions of my input topic -? the state is maintained even when some re-balancing occurs? ---
**Author:** ![Bruno](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/bruno/32/138_2.png) [@Bruno](https://forum.confluent.io/u/Bruno)\ **Post date:** [11 May 2021 06:56 UTC](https://forum.confluent.io/t/running-multiple-aggregations-against-a-topic/1639/4 "2021-05-11T06:56:52Z")
I am not aware of an example. You can find Streams examples under [GitHub - confluentinc/kafka-streams-examples: Demo applications and code examples for Apache Kafka's Streams API.](https://github.com/confluentinc/kafka-streams-examples) In Streams, you can share a state store across multiple operations. See the javadocs for `KStream#transform()` for how to connect a state store: [KStream (kafka 2.8.1 API)](https://kafka.apache.org/28/javadoc/org/apache/kafka/streams/kstream/KStream.html#transform(org.apache.kafka.streams.kstream.TransformerSupplier,java.lang.String...)). Similar applies for `KStream#process()`, `KStream#transformValues()`, `KStream#flatTransform()`, and `KStream#flatTransformValues()`. If your external events can be produced to a Kafka topic, you could read the external events from that topic and process them with one of the operation mentioned above. Let’s assume you use `KStream#process()`. In the `process()` you can set a flag in a state store each time you read an external event from the topic that triggers the interruption of the aggregation. If you connect that same state store to a – let’s say – `KStream#transformValues()` that is put before the aggregation you can read the flag from the shared state store and only pass events to the aggregation if the flag is not set. That is just an idea, there are for sure other methods to achieve what you want. > Also is it safe to assume that I can scale this topology by increasing partitions of my input topic -? the state is maintained even when some re-balancing occurs? Yes, it is safe.