Understanding Akka Streams and Its Components

Table of contents

Reading Time: 4 minutes

Overview

In this blog, we’ll be understand about akka streams and its components. Also, we’ll do a simple exercise that involves each of these components.

Introduction

Stream

A stream is a flow of data that involves moving and transforming data. An element is the processing unit of the stream.

Akka Streams

In software development, there can be cases where we need to handle the potentially large amount of data. So while handling these kinds of scenarios there can be issues such as out of memory exceptions. So we should divide the data in chunks and handle the chunks independently. There come Akka streams for rescue to do this in a more predictable and less chaotic manner.

Akka streams is a module built on top of Akka Actors to make the ingestion and processing of streams easy. It provides easy-to-use APIs to create streams that leverage the power of the Akka toolkit without explicitly defining actor behaviours and messages. This helps you to focus more on logic and forget about all of the boilerplate code required to manage the actor. It is non-blocking that means a certain operation does not hinder the progress of the calling thread, even if it takes a long time to finish the requested operation.

Akka Streams is a library to process and transfer a sequence of elements using bounded buffer space. This latter property is what we refer to as boundedness and it is the defining feature of Akka Streams.

Akka streams consist of three major components in it – Source, Flow and Sink. These are described below in detail.

Components of Akka Streams

Source

Source is an operator that emits data elements asynchronously whenever downstream operators are ready to receive them. It has exactly one output and it’s the entry point to your stream. There must be at least one source in every akka stream. A source may or may not terminate. We can think of Source as a Publisher.

Some of the ways of defining a Source:

val emptySource = Source.empty //An empty source

val singleSource = Source.single(1) //Source emitting a single element

val finiteSource = Source(List(1 to 100)) //Source emitting finite number of elements from a list

val infiniteSource = Source(Stream.from(1)) //Source emitting infinite elements from a collection stream

val futureSource = Source.fromFuture(Future(10)) //Source emitting element from future

Flow

Flow is an operator that takes the elements from the source, apply some transformation in it and transfer it to the Sink. It has exactly one input and output. There can be zero, one or more flows in any akka stream. We can think of Flow as a Processor.

Some of the ways of defining a Flow:

val mapFlow = Flow[Int].map(x => x * 2) //Flow that doubles the incoming elements

val filterFlow = Flow[Int].filter(_ > 5) //Flow taking only elements greater than 5

val takeFlow = Flow[Int].take(5) //Flow taking only 5 elements from the source

Sink

Sink is an operator that receives the data elements from the stream. It has exactly one input and it’s the exit point of your stream. There must be at least one sink in every akka stream. A sink terminates when the source terminates. We can think of Sink as a Receiver or Subscriber.

Some of the ways of defining a Sink:

val foreachSink = Sink.foreach[Int](println) //Sink printing the received elements

val foldSink = Sink.fold[Int,Int](0)((a,b)=> a+b) //Sink adding received elements

val headSink = Sink.head[Int] //Sink receiving only the head element

val ignoreSink = Sink.ignore //Sink not doing anything with the received elements

A Simple Exercise

Let’s do a simple exercise where we’ll create an akka stream that takes names of people from the source, then keeps only two names whose length is greater than 5 characters and then printing them to the console.

import akka.actor.ActorSystem
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.{Flow, Sink, Source}

object SimpleAkkaStreamExercise extends App {

  implicit val system = ActorSystem("SimpleAkkaStreamExercise")
  implicit val materializer = ActorMaterializer()

  val namesSource = Source(List("Prateek", "Ram", "Chaitanya","Amarnath", "Azmat"))
  val filterFlow = Flow[String].filter(name => name.length > 5)
  val takeFlow = Flow[String].take(2)
  val sink = Sink.foreach[String](println)

  namesSource.via(filterFlow).via(takeFlow).to(sink).run() // way of running a akka stream
}