How to minimize memory usage in akka streams

user_440237 0
Viewed 503

I have stream that at some point will group objects to create files. I think I can squeeze some bytes by serializing the object early in the stream. But my biggest question is about how to optimize the memory footprint for a stream like this:

val sourceOfCustomer = Source.repeat(Customer(name = "test"))
def serializeCustomer(customer: Customer) = customer.toString

sourceOfCustomers
.via(serializeCustomer) // 1KB
.grouped(1000000) // 1GB
.via(processFile) // 1GB
.via(moreProcessing) // 1GB
.via(evenMoreProcessing) // 1GB
.to(fileSink) // 1GB

This gives me a memory usage at steady state of at least 5GB. Is this correct?

What strategy can I use to only limit it to 1 or 2GB? In principle it should be possible by collapsing the operators.

Note: I know a solution is to make the group smaller but let’s consider the size of the group a constraint of the problem.

akka-stream akka scala
1 Answers

Sorry, maybe I'm missing something, but I did not find group operation in latest Akka Stream documentation, I guess you mean grouped operation: https://doc.akka.io/docs/akka/current/stream/operators/Source-or-Flow/grouped.html .

If so, then it means that at .grouped(1000000) // 1GB you create group of elements in the stream , which can be handled simultaneously, hence one more then one group of 1GB size can present in memory in one moment of time. So in order to limit memory footprint in your stream up to 1GB, you can go with one of two ways:

1) Reduce number of large groups handled simultaneously. This can be achieved with throttle operation: https://doc.akka.io/docs/akka/current/stream/operators/Source-or-Flow/throttle.html#throttle Please, see code snippet for example

import scala.concurrent.duration._
...

.group(1000000) // 1GB
.throttle(1, 1 minute)

2) Reducing large group size 

val parallelismLevel = Runtime.getRuntime.availableProcessors() // or another custom level which represents stream processing parallelism
val baseGroupSize = 1000000 // 1GB
val groupSize =  baseGroupSize / parallelismLevel 

sourceOfCustomers
.via(serializeCustomer) // 1KB
.group(groupSize)

Hope this helps!

user_32989170
Related