Kafka Streams API: Avoid additional stateStore in KTable.mapValues

user_4667515 0
Viewed 161

Currently we use the following in our Kafka Streams application:

streamsBuilder.table(inputTopic)
              .join(...)
              .mapValues(valueMapper) // <-- this causes another state store
              .groupBy(...)
              .aggregate(...)
              .mapValues(...)

[...]
              .toStream()
              .to(outputTopic)

and I just realized, that the mapValues after the join creates an additional state store.

If the calculation in valueMapper is somehow trivial (e.g. remove a field in an object etc), would it not be better to avoid the additional statestore? Do I need to convert to a KStream and use KStream.mapValues to avoid a stateStore, i.e.

streamsBuilder.table(inputTopic)
              .join(...)
              .toStream
              .mapValues(valueMapper) // <-- no more additional statestore
              .groupBy(...)
              .aggregate(...)
              .mapValues(...)

[...]
              .toStream()
              .to(outputTopic)

or is there a better alternative to apply additonal mapping after a join?

ktable apache-kafka-streams apache-kafka
1 Answers

Why do you want to use a mapValues after a join step? If it is possible to handle that logic in the ValueJoiner of the join.

streamsBuilder.table(inputTopic)
              .join( anotherTable, (a ,b) ->  c )  <--- Here you can perform any mapping process
              .toStream
              .groupBy(...)
              .aggregate(...)
              .mapValues(...)

[...]
              .toStream()
              .to(outputTopic)
user_131918570
Related