Recursive Data Types

I was pointed in the direction of this lecture (and these slides). These are notes I made.

Imagine you want to build a hierarchy of professors and their students who too go on to be professors. You might like to model it like this:

case class ProfF[A](
  name: String,
  year: Int,
  students: List[A]
)

And have case classes to allow us to make recursive types because "type aliases can't be recursive but classes can" (using recursive type aliases result in "types [that] are infinite").

What's more, with ProfF "depending on how we wrap it we can represent just the pure data or the data annotated with database id" thus:

case class Prof(value: ProfF[Prof])

or

case class IdProf(id: Int, value: ProfF[IdProf])

But this is too tightly coupled with ProfF so let's refactor them like so:

case class Prof[F[_]](value: F[Prof])
case class IdProf[F[_]](id: Int, value: F[IdProf])

and too tightly coupled to the type of the id, so let's refactor that too:

case class Prof[F[_]](value: F[Prof[F]])

case class IdProf[F[_], A](id: A, value: F[IdProf[F, A]])

But there is nothing new under the sun as actually these useful data structures already exist in Scalaz.

case class Fix[F[_]](unfix: F[Fix[F]])
case class Cofree[F[_], A](head: A, tail: F[Cofree[F, A]])

and let's also throw in:

case class Free[F[_], A](resume: A \/ F[Free[F, A]])

Which is a type that says resume is either A or F[Free[F, A]]. It's like an Either is Scala. "\/[A, B] is isomorphic to scala.Either[A, B], but \/ is right-biased, so methods such as map and flatMap apply only in the context of the right case" (from here).

"In the same way you think of Cofree as a structure with labels at each node, Free is a structure with labels at the leaves". That is, where Cofree will always have an A, Free can either have an A but no further branches (ie, it's a leaf) or it has branches and no A (ie,  it's a branch to another Free).

Terminology

If F is a functor, Free is the free monad. 

Also, if F is a functor, Cofree is a comonad. More on this later.

"If we have a Fix we can always get an F out by calling unfix, and we can do that with Cofree by calling tail. The general name for this operation is project and data types that do this are called recursive types. And note that we can't do it with Free because you might not have an F; you might have an A. 

"But you can always go the other way. If you have the F you can construct a Free. Same with Fix. But you can't do this with Cofree because you also need an A. So this operation (the dual) is called embed and data types with this behavior are corecursive types."

This is used in the wonderfully named Matryoshka library.

What is it good for?

Well, we can build a recursive data structure to be sent to a persistence layer with this:

type Prof = Fix[ProfF]

and receive a similar (technically corecursive) data structure back but that now contains DB-assigned primary keys with this:

type IdProf = Cofree[ProfF, Int]

These are the basics you need to understand Doobie, a functional library for accessing JDBC.

Spark and Vector Multiplication

More Spark shenanigans. I'm having all sorts of problems while attempting to make the run stable. I'm seeing FetchFailedExceptions that (according to this) are "almost guaranteed to be caused by memory issues on your executors" and FileNotFoundExceptions that mysteriously appear when an executor throws an OutOfMemoryError in the initializer of org.apache.spark.shuffle.sort.ShuffleInMemorySorter.

Thanks for the memory

So, the first thing is to review how Spark manages its memory. The areas are broken down as:

• Reserved Memory
• User Memory
• Spark Memory -broken down into:
• Storage Memory
• Executor Memory

where (from here)

Reserved Memory "is the memory reserved by the system, and its size is hardcoded. As of Spark 1.6.0, its value is 300MB"

User Memory "is the memory pool that remains after the allocation of Spark Memory, and it is completely up to you to use it in a way you like. You can store your own data structures there that would be used in RDD transformations".

Storage Memory "is used for both storing Apache Spark cached data and for temporary space serialized". It also stores broadcast variables.

Execution Memory "is used for storing the objects required during the execution of Spark tasks."


Why heap space isn't the whole story

Here is an example.

Given "a matrix A that has a large number of columns (a short and fat matrix) and a matrix B that has a large number of rows (a tall and thin matrix)" that we want to multiply, then there's a good chance multiplying them may blow up.

The reason is that this can be very memory intensive (see the User Memory section above).

To get around this, we can "apply the outer product to each row of the RDD, a column of A with a row of B. This will create an RDD of small matrices size m*k." We then "use the reduce method to sum the all the small matrices."

Neat. So the trick is to break down each step into smaller steps. In this example, it's done by making an RDD that pairs small vectors together so they can be multiplied with very little overhead.


Refactor the problem

Realizing that my vector multiplication was probably the culprit, I gave myself more user memory but setting spark.memory.fraction much lower (about 0.01) since

User Memory ~ (1.0 - spark.memory.fraction)

This improved things insofar as my run ran for longer but it still ultimately blew up.

The problem was that my sparse matrix was not sparse enough. It could have rows with 2000 non-zero elements. Since the row could have about 100 million elements overall, this is still probably considered sparse. But each element is multiplied with each other element while I calculated Cosine Similarities. This scales as O(N²).

OK, ~2000² results is not the end of the world but what if we get a few of these in the same JVM? We quickly exhaust the memory. Increasing User Memory reduces the likelihood but does not eliminate it. The solution (like the matrix multiplication example above) is to break down the problem into a new RDD that presents the data in smaller chunks.

State monads

State is generally an anathema to functional programmers but necessary to the rest of us. So, how are these forces reconciled? Perhaps not surprisingly it requires monads.

"We'll say that a stateful computation is a function that takes some state and returns a value along with some new state." (from LYAHFGG). So, if you had a random number generator (RNG), for each call you'd receive not just a random number but the generator with a new state for the next call. Calling the same generator repeatedly will give you the same value which makes testing easier.

Scalaz gives you a state out of the box. As an example, let's say we're modelling a stack data structure. The pop would look like this:

import scalaz._
import Scalaz._

    type Stack = List[Int]

    val pop = State[Stack, Int] { // (f: S => (S, A)) where S is Stack and A is Int
      case x::xs =>
        println("Popping...")
        (xs, x)
    }

Note that pop is a val not a def. Like the RNG mentioned earlier, it returns the popped value and the new state. We'll ignore error conditions like popping an empty stack for now.

Also note "the important thing to note is that unlike the general monads we've seen, State specifically wraps functions." (Learning Scalaz)

Let's now define what a push is:

    def push(a: Int) = State[Stack, Unit] {
      case xs =>
        println(s"Pushing $a")
        (a :: xs, ())
    }

Once more, the function returns the new state and a value - only this time the value isn't actually defined. What value can a push to a stack return that you didn't have in the first place?

Because State is a monad, "the powerful part is the fact that we can monadically chain each operations using for syntax without manually passing around the Stack values". That is, we access the contents of State via map and flatMap.

So, let's say we want to pop the value at the top of the stack then push two values onto it, 3 and 4.

    val popx1Push2: State[Stack, Int] = for {
      a <- pop
      _ <- push(3)
      _ <- push(4)
    } yield a 

which does absolutely nothing. What is to be popped? Onto what do we push 3 and 4? We need to run it and then map (or foreach or whatever) over the result to access the values. "Essentially, the reader monad lets us pretend the value is already there." (from Learning Scalaz)
   
    popx1Push2.run(List(0, 1, 2)).foreach { case (state, popped) =>
      println(s"New state = $state, popped value = $popped") // New state = List(4, 3, 1, 2), popped value = 0
    }

But we don't need Scalaz. We can roll our own state (with code liberally stolen from here).

case class State [S, +A](runS: S => (A, S)) {

  def map[B](f: A => B) =
    State [S, B]( s => {
      val (a, s1) = runS(s)
      (f(a), s1)
    })
    
  def flatMap[B](f: A => State [S, B]) =
    State [S, B]( s => {
      val (a, s1) = runS(s)
      f(a).runS(s1)
    })
}

Now it becomes clearer what State actually represents. It is a monad (see map and flatMap) that contains a function that given one state, S, can give you the next plus an accompanying value, A.

So, manipulating the state for our stack example looks like this:

  def getState[S]: State [S, S] =
    State (s => (s, s))
     
where the function says: given a state, return that state.

  def setState[S](s: S): State [S, Unit] =
    State (_ => ((), s))

where the function says: I don't care which state you give me, the next one is s. Oh, and I don't care about the accompanying value.

  def pureState[S, A](a: A): State[S, A] =
    State(s => (a , s))

where the function says the next state is the last state and the accompanying value is what I was given. Pure in this sense refers to the construction of a state, sometimes called unit, and called point in Scalaz.

Now, we do a stateful computation. It's very simple, it just maps over the state monads to generate a unit of work that adds 1 to whatever is given.

    val add1: State[Int, Unit] = for {
      n   <- getState
      b   <- setState(n + 1)
    } yield (b)

    println(add1.runS(7)) // "8"

This is trivial but you can do more complicated operations like:

  def zipWithIndex[A](as: List [A]): List [(Int , A)] =

    as.foldLeft (

      pureState[Int, List[(Int , A)]](List())

    )((acc, a) => for {

      xs  <- acc

      n   <- getState

      _   <- setState(n + 1)

    } yield (n , a) :: xs ).runS(0)._1.reverse

Yoiks! What's this doing? Well, reading it de-sugared and pretending we've given it an array of Chars may make it easier:

    type IndexedChars = List[(Int, Char)]

    val accumulated: State[Int, IndexedChars] = as.foldLeft(

      pureState(List[(Int, Char)]())      // "the next state is the last state and the accompanying value is an empty list"

    )((acc: State[Int, IndexedChars], a: Char) => {

      acc.flatMap { xs: IndexedChars =>   // xs is the accompanying value that the accumulator wraps

        getState.flatMap { n: Int =>      // "Given a state, return that state." (n is fed in via runS)

          setState(n + 1).map(ignored =>  // "I don't care which state you give me, the next one is n+1"

            ((n, a) +: xs)                // the accompanying value is now the old plus the (n,a) tuple.

          )

        }

      }

    })

Which is a bit more understandable.

First steps into Scalaz

Scalaz adds some nice features to Scala.

In the Haskell world, "functions are applicative functors. They allow us to operate on the eventual results of functions as if we already had their results" (from LYAHFGG).

(Where the "Functor typeclass, ... is basically for things that can be mapped over", "all applicatives are functors" [1] and applicatives have a function that has a type of f (a -> b) -> f a -> f b.)

So, why can't we map over functions in Scala? Well, now with Scalaz, we can!

import scalaz._
import Scalaz._

    val f: Int => Int = _ * 2
    val g: Int => Int = _ + 10

    val mapped = f.map(g)
    println(mapped(7))        // "24", ie (7 * 2) + 10

which is the same as saying f.andThen(g) and so the opposite of saying f.compose(g).

We can also flatMap over functions now and as a result can use for comprehensions:

    val addStuff: Int => Int = for {
      a <- f
      b <- g
    } yield (a + b)

    println("addStuff = " + addStuff(7)) // "31"

Why this monadic behaviour is so nice is that "the syntactic sugar of the for-comprehensions abstracts the details nicely enough for the user, who is completely oblivious of the underlying machinery of binding monads" (from here).

We can even raise a plain old Int to the level of monad with:

println(3.point[Id].map(_ + 1)) // prints out "4"

By using point, the "constructor is abstracted out... Scalaz likes the name point instead of pure, and it seems like it’s basically a constructor that takes value A and returns F[A]." (from Learning Scalaz).

[1] Functional Programming in Scala (Chiusano and Bjarnason)

A Spark Quick Win

If you're having trouble making Spark actually work, you're not alone. I can sympathize with the author of these gotchas. Something killing me for weeks was how slow a Spark job was - tens of hours. Further investigation showed lots of GC - something like a third of the run time. Isn't Spark supposed to handle huge volumes of data?

This one tip proved to be a quick-win:

rdd.persist(StorageLevel.MEMORY_AND_DISK_SER)

This meant that the data held in memory was serialized. As a result, the memory footprint was 80% less (we use Kryo). In itself, that's cool. But the great thing was that all our data now fitted into the memory of the cluster.

Sure, Spark can process more memory than is physically available but if it's swapping memory in and out from disk, that's not only slow but it's going to cause a lot of garbage collection.

With this setting, full garbage collections took only 30s for hours of processing.

The downside is that obviously this requires more CPU. But not much more and a lot less than a JVM that's constantly garbage collecting.

Maths and the Principle of Least Surprise

Well, it surprised me.

While Java's maximum and minimum values of Integers really do represent the highest and lowest int values, the same is not true of Doubles. Here, the maximum and minimum values represent the largest and smallest values of the magnitude of the numbers. That is, both values are positive even though Doubles are signed. The nomenclature is confusing but it makes sense (see here). Unlike Integers, Doubles are symmetric so the smallest number you can represent is -Double.MAX_VALUE.

Just when you get used to that, you find that Scala does it differently: Double.MinValue really is the smallest value a double can hold. That is,

Double.MinValue == -java.lang.Double.MAX_VALUE

It's debatable which is the most surprising.

HBase Troubleshooting

Nasty little problem. Connecting to HBase kept timing out after lots of errors that looked like:

client.RpcRetryingCaller: Call exception, tries=17, retries=35, retryTime=208879ms, msg row '' on table 'hbase:meta' at region=hbase:meta,,1.1588230740, hostname=HOST,60020,1475135661112, seqNum=0

while logging in the HBase shell seemed to be OK.

HBase uses Zookeeper to keep a record of its regions. So, first I tried checking that:

> echo ruok | nc ZOOKEEPER_HOST 2181
imok

Running a repair didn't seem to help either.

Time to look at the HBase transaction logs. They kept repeating every second or so something like:

master.splitLogManager: total tasks =3 unassigned = 0 tasks={ /hbase-secure/splitWAL/WALs/HOST,600201474532814834-splitting ... status = in_progess ...

WALs are write ahead logs. "HBase data updates are stored in a place in memory called memstore for fast write. In the event of a region server failure, the contents of the memstore are lost because they have not been saved to disk yet. To prevent data loss in such a scenario, the updates are persisted in a WAL file before they are stored in the memstore".

These WALs appeared to be constantly splitting. So, a quick look at the WALs directory  was in order. It looked something like this:

drwr-x-r-x - hbase app-hdfs     0 2016-07-29 /apps/hbase/xxx/WALs/MACHINE_NAME,60020,1474532814834-splitting
drwr-x-r-x - hbase app-hdfs     0 2016-09-29 /apps/hbase/xxx/WALs/MACHINE_NAME,60020,1474532435621
drwr-x-r-x - hbase app-hdfs     0 2016-09-30 /apps/hbase/xxx/WALs/MACHINE_NAME,60020,1474532365837
drwr-x-r-x - hbase app-hdfs     0 2016-07-28 /apps/hbase/xxx/WALs/MACHINE_NAME,60020,1474532463823-splitting
.
.

Splitting is natural but shouldn't take too long (a split file from months ago is definitely a bad sign).

We were lucky that this was a test environment so we could happily delete these splitting files and restart HBase (YMMV. Think long and hard about doing that in a production environment...) But the problem went away for us.