Architectural and Development Patterns

Event Sourcing
"[E]very operational command executed on any given Aggregate instance in the domain model will publish at least one Domain Event that describes the execution outcome. Each of the events is saved to an Event Store in the order in which it occurred. When each Aggregate is retrieved from its Repository, the instance is reconstituted by playing back the Events in the order in which they previously occurred... To avoid this bottleneck we can apply an optimization the uses Aggregate state snapshots." [1]

CQRS
Instead of having one data store for everything, "the change that CQRS introduces is to split that conceptual model into separate models for update and display, which it refers to as Command and Query respectively". [2]

I've seen this work well in a project that used a Lambda Architecture. Here, risk data was being written slowly but surely to a Hadoop cluster. When a day's data was complete, the data could be converted into a format that was digestible by the Risk Managers.

"Having separate models raises questions about how hard to keep those models consistent, which raises the likelihood of using eventual consistency." [2] For us this was fine as the Risk Managers were in bed while the munging was taking place.

"Interacting with the command-model naturally falls into commands or events, which meshes well with Event Sourcing." [2] And this is where Kafka came in.

Martin Fowler is cautious about using CQRS [2] but we had a good experience. In our domain, the Risk Managers were not interested in the Query model after two weeks and so it could be thrown away. Had it been needed at any time in the future, it could always have been generated again from the Command model.

Lambda Architecture
Lambda Architectures consist of three main pieces:

1. The batch layer, managing the master dataset (an immutable, append-only set of raw data) and pre-computing batch views.
2. The serving layer, indexing batch views so that they can be queried in a low-latency, ad-hoc way.
3. The speed layer, dealing with recent data only, and compensating for the high latency of the batch layer.

(taken from here).

As mentioned in the CQRS subsection above, "even if you were to lose all your serving layer datasets and speed layer datasets, you could reconstruct your application from the master dataset. This is because the batch views served by the serving layer are produced via functions on the master dataset, and since the speed layer is based only on recent data, it can construct itself within a few hours." [3]

Since Hadoop has an append-only, highly available file system (HDFS), it makes it an obvious choice for the batch layer.

"The serving layer is a specialized distributed database that loads in a batch Batch layer view and makes it possible to do random reads on it (see figure 1.9). When new batch views are available, the serving layer automatically swaps those in so that more up-to-date results are available. A serving layer database supports batch updates and random reads. Most notably, it doesn’t need to support random writes." [3] Cassandra was our choice for this layer.

Finally, "the speed layer only looks at recent data, whereas the batch layer looks at all the data at once. Another big difference is that in order to achieve the smallest latencies possible, the speed layer doesn’t look at all the new data at once. Instead, it updates the realtime views as it receives new data instead of recomput­ing the views from scratch like the batch layer does. The speed layer does incre­mental computation instead of the recomputation done in the batch layer." [3]

Lambda architecture shares some similarities with CQRS but is very different in that it never updates data (although it can add data that renders old data obsolete).

Feature Branching
The idea here is that each developer works on his own branch. He regular synchs with the main branch to lessen the pain of later merging his work into it. With Git, he'd regularly run:

> git checkout master
> git pull
> git checkout BRANCH_NAME
> git merge master
> git push

There is some debate on whether Feature Branching is an anti-pattern. Ideally, the system should be architected such that a feature is isolated to a particular silo of the code and not spill out to other components. Thus, a developer can merrily work on that particular silo and check into the main branch all the time. Mainly for political reasons, though, this is not always possible. Software like Stash that integrates with Jira can make code reviews even within a distributed team quite pleasant.

[1] Implementing Domain Driven Design, Vaughn Vernon
[2] Martin Fowler's blog.
[3] Big Data, Nathan Marz

Arrays in Java and Scala

Despite the fact they both run on the JVM, Java and Scala have very different type systems. Say you want to create an array of type T[] in Java where T is any type. It is possible but messy. You might call Array.newInstance but it's signature is:

    public static Object newInstance(Class componentType, int length)

that is, it returns an Object so you'd have to cast it.

If you want to get the component type of a particular array, you'd call Class.getComponentType() but that can be called an any class, not necessarily an array. If it's not called on a type that's an array, it returns null. What's more, it returns Class not Class<T>.

Finally, Java's arrays are covariant (unlike its generics) so you run the danger of getting an ArrayStoreException at runtime if you're not careful. Put another way, Java's arrays don't adhere to the Get and Put Principle that its collections do ("The Get and Put Principle: use an extends wildcard when you only get values out of a structure, use a super wildcard when you only put values into a structure, and don't use a wildcard when both get and put." [1])

Yuck. Scala cleans this up using an invariant Array[T] class constructor. If you JAD the code, you can see how Scala does it. Say we want a multi-dimensional array of Ints, we might call:

    val matrix            = Array.ofDim[Int](7,9)

if we then decompiled the .class file of scala.Array into Java, we can see what Scala is doing. It looks like this:

    public Object[] ofDim(int n1, int n2, ClassTag evidence$4) { ...

where our type (Int) has cheekily been added as an extra parameter. Note, the method returns an array of Object.  As it happens, each of these Objects are arrays of the type we want, in this case Int ultimately using the same Java API method, newInstance, we saw above.

This is another nice feature of Scala in that we can dynamically create arrays of a given type. The ofDim method looks like:

  def ofDim[T: ClassTag](n1: Int): Array[T] = ...

or, equivallently:

  def ofDim(n1: Int)(implicit T: ClassTag[T]): Array[T] = ...

allowing us to do something like:

      val anArray = new Array[T](length) 

without this ClassTag, you'll get "cannot find class tag for element type T".

[1] Java Generics and Collections - Naftalin and Wadler

Future notes (part II)

There's some really nice aspects to Scala's Futures that make them very different to Java's. Because they're monadic, you can do all sorts of interesting things.

Differences between Java and Scala Futures

In Java, you merely block while waiting for the result by calling get(). That's it.

However, you can cancel Java futures. Scala "Futures come without a built-in support for cancellation" [1] although there are ways to roll-your-own.

So what can you do that's so special?

Because both Lists and Futures are monadic, you can "blend" them. For instance, say you have a List of Futures, you can convert them into a Future of Lists. This allows you to react to just a single Future with all the data rather than worrying about many Futures each with some data.

This is built in to the Scala API via Future.sequence. It basically folds over the monadic structure (eg a List) that's passed to it.

A somewhat simplified version of this code (I use the more specific List rather than sequence's more general TraversableOnce and avoids the magic of CanBuildFrom) looks like this:

  def mySequence[T](futures: List[Future[T]])(implicit ec: ExecutionContext): Future[List[T]] = {
    def foldFn(agg: Future[List[T]], ft: Future[T]): Future[List[T]] = {
      agg flatMap { list =>
        ft map {
          element => list :+ element
        }
      }
    }

    val seed: Future[List[T]] = Future {
      List[T]()
    }

    futures.foldLeft (seed) (foldFn)
  }

Or, we could use a for-comprehension to give it some syntactic sugar:

    def foldFn(agg: Future[List[T]], ft: Future[T]): Future[List[T]] = for {
      list <- agg
      t <- ft
    } yield (list :+ t)

They're equivalent. Whether it's this simplified method or the more complicated one in the Scala API, you'd call it with something like:

    val listOfFutures = List(createSimpleFuture("this"), createSimpleFuture("is"), createSimpleFuture("a"), createSimpleFuture("test"))
    val futureOfLists = mySequence(listOfFutures)
    val lists         = Await.result(futureOfLists, Duration(5, TimeUnit.SECONDS))
    println("lists from future of lists from list of futures: " + lists) 


  def createSimpleFuture(eventualResult: String)(implicit ec: ExecutionContext): Future[String] = {
    Future {
      Thread.sleep(100)
      eventualResult
    }
  }

which prints out:

List(this, is, a, test)

More funky stuff in part III.

[1] Learning Concurrent Programming in Scala by Prokopec, Aleksandar - Aleksandar Prokopec

Scala's Return

I was reading this StackOverflow answer about the distinction between methods and functions. It had the usual explanations, for example a method "can be easily converted into [a function]" but "one can't convert the other way around"; methods "can receive type parameters"; "a function is an object" etc.

But as pointed out here, return behaves differently in a function compared to a method and can be used to short-circuit a fold etc.

For example:

  def caller(): Any = {
    println("caller started")
    val g = () => { return true } // note: return type must be compatible with this method
    called(g)
    println("caller finished")    // this is never executed!
  }

  def called(innocuousFn: () => Boolean): Unit = {
    println("called: about to call innocuousFn...")
    innocuousFn()
    println("called: finished")   // this is never executed!
  }

prints out:

caller started
called: about to call innocuousFn...

That is, when the flow of control hits the function's return statement, the execution stack is popped as if it were in the enclosing method (it's illegal for a function outside a method to have a return statement).

Worse, if one method passes a returning function back to the method that calls it, you get a Throwable. After all, where does that function return from?

  def executeReturningFnFromOtherMethod(): Unit = {
    println("about to call calledReturnsFnThatReturns")
    val fn: () => Boolean = returnsFnThatReturns()
    println("about to call fn")
    fn()                // Exception in thread "main" scala.runtime.NonLocalReturnControl
    println("finished") // never executes
  }

  def returnsFnThatReturns(): () => Boolean = {
    val fn: () => Boolean = () => {
      return makeTrue
    }
    fn
  }

  def makeTrue(): Boolean = {
    return true
  }

Prints out:

about to call calledReturnsFnThatReturns

about to call fn

Exception in thread "main" scala.runtime.NonLocalReturnControl

Methods that may or may not compile in Scala

If you want to sum some integers in a List, you just call the sum method. So far, so obvious. But if the list does not hold elements that are summable, how does Scala give you a method that you may or may not call?

The sum method lives in TraversableOnce and the signature looks like this:

  def sum[B >: A](implicit num: Numeric[B]): B = ...

Here [B >: A] says B must be a superclass of A, the type of elements in the list. The implicit says there must exist in the ether something that provides the functionality in trait Numeric for type B (the functionality for plus, minus etc).

Now, for Int, Double etc you get these for free in the Numeric object that's standard to Scala where you see, for example:

  trait IntIsIntegral extends Integral[Int] { ...

which is pulled into the implicit ether with:

  implicit object IntIsIntegral extends IntIsIntegral with Ordering.IntOrdering

in the same file. Note, that unlike Java, Scala's numerics do not extend a common superclass like java.lang.Number.

So, we could happily create our own arithmetic with something like:

  class MyInteger

  trait MyIntegerOrdering extends Ordering[MyInteger] {

    def compare(x: MyInteger, y: MyInteger) = ???

  }

  trait MyIntegral extends Numeric[MyInteger] { // or Integral that extends Numeric[T]

    def plus(x: MyInteger, y: MyInteger): MyInteger = new MyInteger

.

.

.

  }

  implicit object MyIntegerIsIntegral extends MyIntegral with MyIntegerOrdering

It's not a terribly useful arithmetic but it would allow us to have a list of MyIntegers and call sum on it.

Notes on the Future (part 1)

Scala's Futures are very different to Java's. Where Java's Futures encourage blocking by their API, Scala's encourages asynchronous coding... apart from one gotcha that follows shortly.

Futures are Monads so they have all that Monad goodness and we can use them in for comprehensions. Unfortunately, this very idiom can introduce a bug. Take this code:

for {
  i <- Future(5)
  j <- Future(11)
} yield j

It creates two futures operating that return the values 5 and 11, possibly on different threads, and returns a Future that contains 11 just like a good Monad should.

"Here, the beauty of the for comprehension sugar can work against us. The for comprehension hides all of that ugly nesting from us, making it look like these are on the same level, but nesting is indeed happening here. What might look concurrent isn't." [1]

Gotcha

What happens here is the first Future is executed and only then is the second. The solution is simple. The code should look like this:

val iFuture = Future(5)
val jFuture = Future(11)
for {
  i <- iFuture
  j <- jFuture
} yield j

For just returning a number, you probably won't notice anything amiss. But for something more ambitious, you may very well notice that the first cut of this code is sub-optimal.

"Only men's minds could have unmapped into abstraction..."

In Scala, we can get our dirty little hands on the contents of a Future by mapping like:

import scala.concurrent.ExecutionContext

    implicit val xc = ExecutionContext.global // needed for next map:

    j map { value =>
      println("j = " + value)
    }

but there are other ways, too. Being monads, we can flatMap and so if we slightly modify the code in [1]:

  implicit val xc = ExecutionContext.global

  val timeToWait = Duration(5, TimeUnit.SECONDS)

  val future1 = Future[Int] {
    (1 to 3).foldLeft(0) { (a, i) =>
      log("future1 "  + i)
      pause()
      a + i
    }
  }

  val future2 = Future[String] {
    ('A' to 'C').foldLeft("") { (a, c) =>
      log("future2 " + c)
      pause()
      a + c
    }
  }

  def spliceFutures(): Unit = {
    val awaitable = future1 flatMap { numsum: Int =>
      log("in future1.flatMap") // ForkJoinPool-1-worker-1: in future1.flatMap

      future2 map { str: String =>
        log("in future2.map") // ForkJoinPool-1-worker-5: in future2.map
        (numsum, str)
      }
    }

    val awaited = Await.result(awaitable, timeToWait)
    println(awaited) // (6,ABC)
  }

  def log(msg: String): Unit = {
    println(Thread.currentThread().getName() + ": " + msg)
  }

"The most powerful way to use these combinators is in combining the results from futures that are already executing. This means that we can still have a lot of independent futures running, and have another piece of code combine the eventual results from those futures into another "final" future that operates on those results."[1]

You'll notice from the log statements that the code in the map and flatMap run in different threads.

But wait, there's more. You can add callbacks with onSuccess, onFailure and onComplete, all of which have return type of Unit (so you know that the functions passed to it must have side-effects) and there is no guarantee that they'll run on the same thread that ran the body of the Future.

  val onCompleteFn: Try[Boolean] => Unit = {

    case Success(x)   => println(" success: " + x)

    case Failure(err) => println(" failure: " + err + ", " + err.getCause)

  }

.

.

    aFuture onComplete onCompleteFn 

In fact, Future.onComplete is the daddy. All the other combinator methods (map, flatMap, recover, transform etc) all call it. We'll come back to these other combinators in another post.

Problems in your Future

If an Exception is thrown in your Future, then it will be captured in the failure case in onCompleteFn above. But if an Error is thrown, you'll get a Boxed Error.

Another surprise comes from filter. For Lists, Trys and Options, things are quite straightforward:

    val myList = List(1,2,3)
    val filteredList = myList.filter(_ > 10)
    println(filteredList)   // List()

    val myOption = Option(1)
    val filteredOption = myOption.filter(_ > 10)
    println(filteredOption) // None

    val myTry = Try(1)
    val filteredTry = myTry.filter(_ > 10)
    println(filteredTry)    // Failure(java.util.NoSuchElementException: Predicate does not hold for 1)

That is, a filter whose predicate leads to nothing is represented by the "negative" subclass of the monads. However, there is no such class for an "empty" Future.

    import scala.concurrent.ExecutionContext.Implicits._

    import scala.concurrent.duration._

    val myFuture       = Future { 1 }

    val filteredFuture = myFuture filter {_ > 10}

    filteredFuture onComplete {

      case Success(x) => println(s"success: $x")

      case Failure(x) => println(s"failure: $x")

    } // failure: java.util.NoSuchElementException: Future.filter predicate is not satisfied

    Await.result(filteredFuture, 1.second) // Exception in thread "main" java.util.NoSuchElementException: Future.filter predicate is not satisfied

That is, the Failure case is the path taken by onComplete. If we were so foolish to block using Await, then an Exception is thrown.

[1] Akka Concurrency, Derek Wyatt.

Spark Shuffle

... is something you should avoid.  "All shuffle data must be written to disk and then transferred over the network" (from here). This can be very expensive on a large data set. If the data is evenly distributed but a query asks for it to be grouped according to an arbitrary criteria, much movement will ensue. Therefore, groupByKey should be avoided if possible as the ScalaDocs indicate this can be an expensive operation.

However, reduceByKey is much more efficient as it will "perform the merging locally on each mapper before sending results to a reducer" resulting in a much smaller transfer of data between nodes.

So, when we were writing a risk engine with Spark holding all our user-defined Record objects, our code would look something like:

def totalByKey(data: RDD[Record]): RDD[(String, Double)] = {
  data.map(record => record.key -> record.value).reduceByKey(_ + _)
}

which would first turn our Records into key/value pairs then add up all the values for each key without transferring large amounts of data over the network. It avoids this by adding all the values for Records on each node before adding all these subtotals. Only the addition of subtotals requires a network call.

Interestingly, Spark can then do an outer join like this:

val subtotalForDate1 = totalByKey(dataForDate1)

val subtotalForDate2 = totalByKey(dataForDate2)

val outerJoing       = subtotalForDate1.fullOuterJoin(subtotalForDate2)

which returns an RDD of tuples containing the key, the left and the right value. Using this join, we can compare our risk profile from one date to the next.

Note that both reduceByKey and fullOuterJoin don't actually live in the RDD class but the PairRDDFunctions class. They appear to be part of the RDD class by the magic of implicits when the RDD in question contains tuples of pairs.

One last thing to remember is that if you're trying to be clever and reduce collections of elements, you're no better off than using groupByKey. The function reduceByKey is only useful if you are reducing something like a number. If you look at the code for groupByKey you'll see that it's basically doing just that: reducing collections.

Further Reading

Nice article on memory usage.