How to implement batching logic in Akka Streams

How to implement batching logic in Akka Streams

A common request we see with streaming data is the need to take the stream of elements and group them together (i.e. committing data to a database, a message queue or disk). Batching is usually a more efficient and performant solution than writing a single piece of data at a time.

Using the Akka Streams API, grouping messages is as easy as adding a grouped element.
  implicit val system = ActorSystem()
  implicit val materializer = ActorMaterializer()
  implicit val ec = system.dispatcher

  val groupedExample =
    Source(1 to 100000)
      .grouped(100)
      .runForeach(println)
      .onComplete(_ => system.terminate())
However, grouping often introduces an unacceptable latency. To address this, you can use the groupedWithin method to group elements within a bounded time frame. This operation takes two parameters, a maximum batch size and a batch cutoff time, which are used to batch together either the specified number of elements or as many elements as are received during the specified duration. Even if the maximum number of elements has not been satisfied, once the specified duration is reached the current grouping will be emitted.
   implicit val system = ActorSystem()
   implicit val materializer = ActorMaterializer()
   implicit val ec = system.dispatcher
 
   val groupedWithinExample =
     Source(1 to 100000)
       .groupedWithin(100, 100.millis)
       .map(elements  => s"Processing ${elements.size} elements")
       .runForeach(println)
       .onComplete(_ => system.terminate())

For grouped - See more in the Java or Scala documentation.

For groupedWithin - See more in the Java or the Scala documentation.


    • Related Articles

    • How to do Rate Limiting in Akka Streams

      In certain scenarios it is important to limit the number of concurrent requests to other services. For example, to avoid overwhelming the services and avoid performance degradation, or to maintain service level agreements. This is particularly ...

    • How to do Throttling in Akka Streams

      When building a streaming application you may find the need to throttle the upstream so as to avoid exceeding a specified rate. Akka Stream's provides the capability to either fail the stream or shape it by applying back pressure. This is simply done ...

    • Error handling and recovery in Akka Streams

      When developing applications you should assume that there will be unexpected issues. For this, Akka provides a set of supervision strategies to deal with errors within your actors. Akka streams is no different, in fact its error handling strategies ...

    • Terminating a stream

      Streams do not run on the caller thread, instead they run on a different background thread. This is done to avoid blocking the caller. Therefore, once the stream completes, you need to terminate the underlying actor system to completely end the ...

    • How to stream multiple stream actions concurrently

      To construct efficient, scalable and low-latency data streams, it is often important to perform tasks concurrently. However Akka Streams executes stages sequentially on a single thread, so you must explicitly request this concurrency. Inserting these ...