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.