Quick Start Guide: Reactive Tweets
Loading

Quick Start Guide: Reactive Tweets

A typical use case for stream processing is consuming a live stream of data that we want to extract or aggregate some other data from. In this example we'll consider consuming a stream of tweets and extracting information concerning Akka from them.

We will also consider the problem inherent to all non-blocking streaming solutions: "What if the subscriber is too slow to consume the live stream of data?". Traditionally the solution is often to buffer the elements, but this can—and usually will—cause eventual buffer overflows and instability of such systems. Instead Akka Streams depend on internal backpressure signals that allow to control what should happen in such scenarios.

Here's the data model we'll be working with throughout the quickstart examples:

final case class Author(handle: String)
val AkkaTeam = Author("akkateam")
val Akka = Hashtag("#akka")

final case class Hashtag(name: String)

final case class Tweet(author: Author, timestamp: Long, body: String) {
  def hashtags: Set[Hashtag] =
    body.split(" ").collect { case t if t.startsWith("#") => Hashtag(t) }.toSet
}

Transforming and consuming simple streams

In order to prepare our environment by creating an ActorSystem and FlowMaterializer, which will be responsible for materializing and running the streams we are about to create:

implicit val system = ActorSystem("reactive-tweets")
implicit val mat = FlowMaterializer()

The FlowMaterializer can optionally take MaterializerSettings which can be used to define materialization properties, such as default buffer sizes (see also stream-buffering-explained-scala), the dispatcher to be used by the pipeline etc. These can be overridden on an element-by-element basis or for an entire section, but this will be discussed in depth in stream-section-configuration.

Let's assume we have a stream of tweets readily available, in Akka this is expressed as a Source[Out]:

val tweets: Source[Tweet]

Streams always start flowing from a Source[Out] then can continue through Flow[In,Out] elements or more advanced graph elements to finally be consumed by a Sink[In]. Both Sources and Flows provide stream operations that can be used to transform the flowing data, a Sink however does not since its the "end of stream" and its behavior depends on the type of Sink used.

In our case let's say we want to find all twitter handles of users which tweet about #akka, the operations should look familiar to anyone who has used the Scala Collections library, however they operate on streams and not collections of data:

val authors: Source[Author] =
  tweets
    .filter(_.hashtags.contains(Akka))
    .map(_.author)

Finally in order to materialize and run the stream computation we need to attach the Flow to a Sink[T] that will get the flow running. The simplest way to do this is to call runWith(sink) on a Source[Out]. For convenience a number of common Sinks are predefined and collected as methods on the :class:Sink companion object. For now let's simply print each author:

authors.runWith(Sink.foreach(println))

or by using the shorthand version (which are defined only for the most popular sinks such as FoldSink and ForeachSink):

authors.foreach(println)

Materializing and running a stream always requires a FlowMaterializer to be in implicit scope (or passed in explicitly, like this: .run(mat)).

Flattening sequences in streams

In the previous section we were working on 1:1 relationships of elements which is the most common case, but sometimes we might want to map from one element to a number of elements and receive a "flattened" stream, similarly like flatMap works on Scala Collections. In order to get a flattened stream of hashtags from our stream of tweets we can use the mapConcat combinator:

val hashtags: Source[Hashtag] = tweets.mapConcat(_.hashtags.toList)

Note

The name flatMap was consciously avoided due to its proximity with for-comprehensions and monadic composition. It is problematic for two reasons: firstly, flattening by concatenation is often undesirable in bounded stream processing due to the risk of deadlock (with merge being the preferred strategy), and secondly, the monad laws would not hold for our implementation of flatMap (due to the liveness issues).

Please note that the mapConcat requires the supplied function to return a strict collection (f:Out=>immutable.Seq[T]), whereas flatMap would have to operate on streams all the way through.

Broadcasting a stream

Now let's say we want to persist all hashtags, as well as all author names from this one live stream. For example we'd like to write all author handles into one file, and all hashtags into another file on disk. This means we have to split the source stream into 2 streams which will handle the writing to these different files.

Elements that can be used to form such "fan-out" (or "fan-in") structures are referred to as "junctions" in Akka Streams. One of these that we'll be using in this example is called Broadcast, and it simply emits elements from its input port to all of its output ports.

Akka Streams intentionally separate the linear stream structures (Flows) from the non-linear, branching ones (FlowGraphs) in order to offer the most convenient API for both of these cases. Graphs can express arbitrarily complex stream setups at the expense of not reading as familiarly as collection transformations. It is also possible to wrap complex computation graphs as Flows, Sinks or Sources, which will be explained in detail in Constructing Sources, Sinks and Flows from a Partial Graphs. FlowGraphs are constructed like this:

val writeAuthors: Sink[Author] = ???
val writeHashtags: Sink[Hashtag] = ???
val g = FlowGraph { implicit builder =>
  import FlowGraphImplicits._

  val b = Broadcast[Tweet]
  tweets ~> b ~> Flow[Tweet].map(_.author) ~> writeAuthors
            b ~> Flow[Tweet].mapConcat(_.hashtags.toList) ~> writeHashtags
}
g.run()

Note

The ~> (read as "edge", "via" or "to") operator is only available if FlowGraphImplicits._ are imported. Without this import you can still construct graphs using the builder.addEdge(from,[through,]to) method.

As you can see, inside the FlowGraph we use an implicit graph builder to mutably construct the graph using the ~> "edge operator" (also read as "connect" or "via" or "to"). Once we have the FlowGraph in the value g it is immutable, thread-safe, and freely shareable. A graph can can be run() directly - assuming all ports (sinks/sources) within a flow have been connected properly. It is possible to construct PartialFlowGraph s where this is not required but this will be covered in detail in Constructing and combining Partial Flow Graphs.

As all Akka Streams elements, Broadcast will properly propagate back-pressure to its upstream element.

Back-pressure in action

One of the main advantages of Akka Streams is that they always propagate back-pressure information from stream Sinks (Subscribers) to their Sources (Publishers). It is not an optional feature, and is enabled at all times. To learn more about the back-pressure protocol used by Akka Streams and all other Reactive Streams compatible implementations read Back-pressure explained.

A typical problem applications (not using Akka Streams) like this often face is that they are unable to process the incoming data fast enough, either temporarily or by design, and will start buffering incoming data until there's no more space to buffer, resulting in either OutOfMemoryError s or other severe degradations of service responsiveness. With Akka Streams buffering can and must be handled explicitly. For example, if we are only interested in the "most recent tweets, with a buffer of 10 elements" this can be expressed using the buffer element:

tweets
  .buffer(10, OverflowStrategy.dropHead)
  .map(slowComputation)
  .runWith(Sink.ignore)

The buffer element takes an explicit and required OverflowStrategy, which defines how the buffer should react when it receives another element element while it is full. Strategies provided include dropping the oldest element (dropHead), dropping the entire buffer, signalling errors etc. Be sure to pick and choose the strategy that fits your use case best.

Materialized values

So far we've been only processing data using Flows and consuming it into some kind of external Sink - be it by printing values or storing them in some external system. However sometimes we may be interested in some value that can be obtained from the materialized processing pipeline. For example, we want to know how many tweets we have processed. While this question is not as obvious to give an answer to in case of an infinite stream of tweets (one way to answer this question in a streaming setting would to create a stream of counts described as "up until now, we've processed N tweets"), but in general it is possible to deal with finite streams and come up with a nice result such as a total count of elements.

First, let's write such an element counter using FoldSink and then we'll see how it is possible to obtain materialized values from a MaterializedMap which is returned by materializing an Akka stream. We'll split execution into multiple lines for the sake of explaining the concepts of Materializable elements and MaterializedType

    val sumSink = Sink.fold[Int, Int](0)(_ + _)

    val counter: RunnableFlow = tweets.map(t => 1).to(sumSink)
    val map: MaterializedMap = counter.run()

    val sum: Future[Int] = map.get(sumSink)

    sum.map { c => println(s"Total tweets processed: $c") }
  }

}

First, we prepare the FoldSink which will be used to sum all Int elements of the stream. Next we connect the tweets stream though a map step which converts each tweet into the number 1, finally we connect the flow to the previously prepared Sink. Notice that this step does not yet materialize the processing pipeline, it merely prepares the description of the Flow, which is now connected to a Sink, and therefore can be run(), as indicated by its type: RunnableFlow. Next we call run() which uses the implicit FlowMaterializer to materialize and run the flow. The value returned by calling run() on a RunnableFlow or FlowGraph is MaterializedMap, which can be used to retrieve materialized values from the running stream.

In order to extract an materialized value from a running stream it is possible to call get(Materializable) on a materialized map obtained from materializing a flow or graph. Since FoldSink implements Materializable and implements the MaterializedType as Future[Int] we can use it to obtain the Future which when completed will contain the total length of our tweets stream. In case of the stream failing, this future would complete with a Failure.

The reason we have to get the value out from the materialized map, is because a RunnableFlow may be reused and materialized multiple times, because it is just the "blueprint" of the stream. This means that if we materialize a stream, for example one that consumes a live stream of tweets within a minute, the materialized values for those two materializations will be different, as illustrated by this example:

val sumSink = Sink.fold[Int, Int](0)(_ + _)
val counterRunnableFlow: RunnableFlow =
  tweetsInMinuteFromNow
    .filter(_.hashtags contains Akka)
    .map(t => 1)
    .to(sumSink)

// materialize the stream once in the morning
val morningMaterialized = counterRunnableFlow.run()
// and once in the evening, reusing the
val eveningMaterialized = counterRunnableFlow.run()

// the sumSink materialized two different futures
// we use it as key to get the materialized value out of the materialized map
val morningTweetsCount: Future[Int] = morningMaterialized.get(sumSink)
val eveningTweetsCount: Future[Int] = eveningMaterialized.get(sumSink)

Many elements in Akka Streams provide materialized values which can be used for obtaining either results of computation or steering these elements which will be discussed in detail in Stream Materialization. Summing up this section, now we know what happens behind the scenes when we run this one-liner, which is equivalent to the multi line version above:

val sum: Future[Int] = tweets.map(t => 1).runWith(sumSink)

Contents