Spark and serialization

One of the trickiest issues for Spark developers is serialization. I'm going to provide a checklist of things to think about when your Spark job doesn't work as expected. This article is directed at Scala programmers, although the principles are the same of course for Java developers (with the exception that there are no Case Classes in Java).

Serialization of functions

This is the one that newbies run into pretty quickly. Any function that you pass to one of Spark's higher-order functions (such as the map method of RDD) must be serializable. Once you think about it, it's pretty obvious but when you're new to Spark, it may not be so clear. All of the real work of Spark is performed on the executors, and so any such function has to be sent "over the wire" to the executors. Hence, the function must be serializable. Functions will be serializable in any of the following three situations:

• the function is a "pure" function, for example one that is created as an anonymous function (i.e. a lambda);
• the function is a partially applied method (that's to say the method has undergone the 𝜂 (eta) transformation whereby the method is followed by the underscore character) where the method is part of:
• an object
• a serializable class

Note what's missing: the function derives from a method belonging to a non-serializable class. This is because an instance function (i.e. belonging to a class rather than an object) has an (implicit) parameter for this, where this is the appropriate instance of the class.

UDFs and Broadcasts

Don't forget, too, that anything object which is passed as a parameter into a class representing a UDF (user-defined function), or anything which you broadcast, must also be serializable.

Abstract Classes

Any class whose instances must be serialized (including, as described above, classes where the methods are used as functions) needs to be marked Serializable. But: even if you mark your concrete class as Serializable, you may still run into a problem with serialization. In such a case, you must ensure that any abstract classes which the class extends are also marked Serializable.

Case Classes

Case classes (and case objects) in Scala are inherently serializable. This is a good thing but it can have unforeseen consequences if your case class is mutable. Let's say you create a case class which has a mutable field (or variable). Perhaps there is a method which can be called to initialize that field or variable. When you run it in the driver, everything looks good. But, if that object needs to be serialized, then the initialization method won't be called after deserialization. If you had had to explicitly mark the case class as serializable, you might have thought about this problem beforehand.

Preemptive serialization checking

If you want to check that objects/functions are serializable before you get too far into your Spark job -- or perhaps you are performing a functional test, then you can use the following Spark environment variable:

SparkEnv.get.closureSerializer

(this is the same serializer that Spark will use). Here's a handy case class/method which will validate serialization:

case class Serialization(serializer: Serializer) {
  def validate[T : ClassTag](obj: T): Boolean = {
    val serializerInstance = serializer.newInstance()
    val result: T = serializerInstance.deserialize(serializerInstance.serialize(obj))
    result==obj
  }
}

Just pass the result of  invoking SparkEnv.get.closureSerializer into the Serialization class and run the validate method on anything you want to check. If it returns true, then all is well. If it returns false, then the deserialized version isn't the same as the original (e.g. you tried to serialize a mutable class as described above). If it throws an exception, then something in the serialization stack (essentially the relevant class hierarchy) isn't marked Serializable when it should be.

Render unto Caesar...

There's a minor issue with both Java and Scala, at least Scala out-of-the-box. That is, while there is a way to turn any object into a String for debugging purposes (it's the toString method available in both Java and Scala), there is no easy, consistent way to turn an object into a String that can be used for presentation.

In a former life, I have created methods to do this for Java. But it's frustrating at best. In Java, if you want a class to take on some particular behavior, you have to arrange for that class to extend an interface that defines the behavior. If that class, some sort of collection for instance, isn't under your control, you're out of luck.

Not so in Scala!  You can add behavior to a (someone else's) library class through the magic of implicits.

So, let's say we define the presentation version of toString in a trait called Renderable, with one method render.

trait Renderable {
  def render: String
}

Any class that you want to be able to prettify in output can extend Renderable. So far, so good. But suppose that you have a list of Renderable objects. It's not much use if you can make the elements pretty but not the whole list. You could, for example, define a RenderableSeq as follows:

case class RenderableSeq[A <: Renderable](as: Seq[A]) extends Renderable {
  def render: String = as map (_ render) mkString ","
}

But, apart from the fact that this is rather ugly and involves creating a set of special containers with which to wrap your sequences, it's simply impractical. Much of the time, you will be dealing with containers that are elements of other classes and you won't be able to inject your desired behavior.

This is where implicits come to the rescue.

import scala.util._

trait Renderable {
  def render: String
}

case class RenderableTraversable(xs: Traversable[_]) extends Renderable {
  def render: String = {
    def addString(b: StringBuilder, start: String, sep: String, end: String): StringBuilder = {
      var first = true
      b append start
      for (x <- xs) {
        if (first) {
          b append Renderable.renderElem(x)
          first = false
        }
        else {
          b append sep
          b append Renderable.renderElem(x)
        }
      }
      b append end
      b
    }
    addString(new StringBuilder, "(\n", ",\n", "\n" + ")").toString()
  }
}

case class RenderableOption(xo: Option[_]) extends Renderable {
  def render: String = xo match {
    case Some(x) => s"Some(" + Renderable.renderElem(x) + ")"
    case _ => "None"
  }
}

case class RenderableTry(xy: Try[_]) extends Renderable {
  def render: String = xy match {
    case Success(x) => s"Success(" + Renderable.renderElem(x) + ")"
    case _ => "Failure"
  }
}

case class RenderableEither(e: Either[_, _]) extends Renderable {
  def render: String = e match {
    case Left(x) => s"Left(" + Renderable.renderElem(x) + ")"
    case Right(x) => s"Right(" + Renderable.renderElem(x) + ")"
  }
}

object Renderable {
  implicit def renderableTraversable(xs: Traversable[_]): Renderable = RenderableTraversable(xs)

  implicit def renderableOption(xo: Option[_]): Renderable = RenderableOption(xo)

  implicit def renderableTry(xy: Try[_]): Renderable = RenderableTry(xy)

  implicit def renderableEither(e: Either[_, _]): Renderable = RenderableEither(e)

  def renderElem(elem: Any): String = elem match {
    case xo: Option[_] => renderableOption(xo).render
    case xs: Traversable[_] => renderableTraversable(xs).render
    case xy: Try[_] => renderableTry(xy).render
    case e: Either[_, _] => renderableEither(e).render
    case r: Renderable => r.render
    case x => x.toString
  }
}

With this design, you can simply invoke render on any object that is either Renderable itself or one of the four implicit Renderable containers defined above. A complex class with embedded options, sequences, whatever simply has to extend Renderable and all elements that can be rendered thus will be (other classes will simply be converted to String via toString). Your application code hardly needs to be aware of the Renderable trait because the implicit converters are defined, implicitly, in the Renderable companion object.

In practice, this particular render method isn't all that useful. I defined it thus to simplify the logic of the implicit definitions. In my LaScala repository on Github [currently only defined in branch V_1_0_1 but soon to be merged with the master branch], I have defined a rather more sophisticated version as follows:

trait Renderable {

  /**
    * Method to render this object in a human-legible manner.
    *
    * @param indent the number of "tabs" before output should start (when writing on a new line).
    * @param tab    an implicit function to translate the tab number (i.e. indent) to a String of white space.
    *               Typically (and by default) this will be uniform. But you're free to set up a series of tabs
    *               like on an old typewriter where the spacing is non-uniform.
    * @return a String that, if it has embedded newlines, will follow each newline with (possibly empty) white space,
    *         which is then followed by some human-legible rendition of *this*.
    */
  def render(indent: Int = 0)(implicit tab: Int => Prefix): String
}
case class Prefix(s: String) {
  override def toString: String = s
}

object Prefix {

  /**
    * The default tab method which translates indent uniformly into that number of double-spaces.
    *
    * @param indent the number of tabs before output should start on a new line.
    * @return a String of white space.
    */
  implicit def tab(indent: Int): Prefix = Prefix(" " * indent * 2)
}

The other classes are more or less exactly as shown above, but with the more complex version of render supported. It would be possible to add further complexity, for example letting the render method choose whether to use newlines rather than commas--according to the space available. But so far, I haven't felt that the extra complexity was really justified.

Some mistakes to avoid when using Scalatest

I've been writing a lot of Scalatest specifications lately. And I've discovered a couple of tips which I think are worth sharing.

First, it's a common pattern, especially if you are doing anything like parsing, that you want to do something like this:

val expr = "x>1 & (x<3 | x=99)""rule" should s"parse $expr using parseRule" in {
  val parser = new RuleParser()
  parser.parseRule(expr) should matchPattern { case scala.util.Success(_) => }
}

Unfortunately, there's a snag. If you want to repeat that specific test (and not the remainder of the Specification), it will result in a run that is green (!) but which contains the message "Empty test suite". That seems like a bug in the test runner to me--it shouldn't show green when it didn't actually test anything. But there's an easy workaround for this that will allow you to repeat that specific test. Just write it like this:

val expr = "x>1 & (x<3 | x=99)""rule" should "parse " + expr + " using parseRule" in {
  val parser = new RuleParser()
  parser.parseRule(expr) should matchPattern { case scala.util.Success(_) => }
}

It's very slightly more awkward to write the code but it's not that hard, especially if you know the trick to splitting a String. Just be sure to avoid using s"...".The second problem I ran into is a little more troubling.

behavior of "silly"
it should "use for comprehension properly" in {
  for (x <- Try("x")) yield x=="y" should matchPattern { case Success(true) => }
}

The result of running this was green:
Testing started at 11:06 AM ...

Process finished with exit code 0

See anything wrong? Clearly, the result should have been Success(false) but it apparently matched. Or did it? Actually, it the whole expression of yield through } was run as a side effect, yielding nothing. And, although I haven't really gone into detail, clearly the test runner didn't get any message about a failure. I don't think this is a bug, although it might perhaps be possible to fix it. But all you have to do is to ensure that you put parentheses around the entire for comprehension thus:

(for (x <- Try("x")) yield x=="y") should matchPattern { case Success(true) => }

Get considered harmful

In Scala, there are several monadic containers for which the get method can throw an exception. The two most obvious such containers are Option and Try. If you have an instance of Option[T] and you invoke get, you will get either a T or throw a java.util.NoSuchElementException; if you have an instance of Try[T] and you invoke get, you will get either a T or throw the exception. Future[T] doesn't have a get method but it has value, which yields Option[Try[T]] which has two potential exception-throwing get methods.

So, we learn that using get is bad: a code smell. Idiomatic Scala recommends using pattern matching or monadic operations like map, flatMap, etc. for working with values from such containers.

So, if get is so bad, then why is Map[K,V] designed the way it is? Map[K,V] extends GenMapLike[K,V], which actually defines the following two methods:

• abstract def apply(key: K): V
• abstract def get(key: K): Option[V]

Thus, effectively, GenMapLike[K,V] extends Function1[K,V]. This class was originally written by Martin Odersky, although not until 2.9. I'm assuming that it was written this way to be compatible with earlier versions. In Scala 2.8, MapLike[K,V] extends Function1[K,V] via PartialFunction[K,V]. Again, Odersky was the author.

If the value of key is unknown in the map, the so-called default method is invoked. By default, the default method simply throws an exception (as you'd expect). In other words, apply is the dangerous method (default notwithstanding) and get is the safe method.

So, is it just me, or are these definitions backwards? This is how I feel that the methods in GenMapLike[K,V] should be defined:

• abstract def get(key: K): V
• abstract def apply(key: K): Option[V]

Thus, GenMapLike[K,V] would effectively extend Function1[K,Option[V]]. What would be wrong with that? It would be so much more consistent. By all means have this version of get invoke the default mechanism. But it would still essentially correspond to the dangerous method.

Obviously, nobody is going to change it now. And, they're just names, right? But it does seem a shame that it got this way in the first place.

Scala's pesky "contravariant position" problem

I'm sure this has happened to you. You design a data structure, for example a tree and, because you are thinking essentially of using this tree to look things up (perhaps it's an AST for example) you decide to make the tree covariant in its underlying type:

trait Tree[+A] extends Node[A] {
   def children: Seq[Node[A]]
}

Now, it's time to start building this tree up from nodes and branches so you add a method like the following:

  def +:(a: A): Tree[A] = :+(a)

Aargh! it's the dreaded "Covariant type A appears in contravariant position in type A of value a." Does this mean that we have to re-write our Tree trait so that it is invariant (as we would have to do if tree was a mutable data structure like Array)? Not at all.

There is a simple and pretty much automatic way of refactoring the +: method. Use the fact that you can define a parametric type at the method level and constrain it to be a super-type of A. Thus:

def +:[B >: A](a: B): Tree[B] = :+(a)

Since B is now a super-type of A (or, of course, A itself), then we declare a (we could rename it b now of course) as a B and we return a Tree[B]. Problem solved.

Note also that if the type of a needs to have one or more context bounds you can declare these too, just as you would expect:

  def +:[B >: A : TreeBuilder](b: B): Tree[B] = :+(b)

I should perhaps point out that this problem is not unique to Scala. It will occur (or should occur) in any object-oriented language that allows generic typing (or anywhere the Liskov substitution principle applies). It's only because Java, for example, glosses over the whole concept of variance, that we don't find the problem there too.

Next time I will talk about having more than one context bound for a type.

Spying -- or a functional way to log in Scala

All of the logging utilities I've found for Scala suffer from the same problem: they are not written in a functional-programming style. Here's the problem: you want to log the result of a function which, to choose an example more or less at random, looks like this:

  def map2[T, U](to1: Option[T], to2: => Option[T])(f: (T, T) => U): Option[U] = for {t1 <- to1; t2 <- to2} yield f(t1, t2)

Now, we decide that we'd like to log the result of this method (or maybe we just want to print it to the console). To do this the "traditional" way, we would need to write something like this:

  def map2[T, U](to1: Option[T], to2: => Option[T])(f: (T, T) => U): Option[U] = {
    val logger = LoggerFactory.getLogger(getClass)
    val r = for {t1 <- to1; t2 <- to2} yield f(t1, t2)
    logger.debug(s"map2: ${r.toString}")
    r
  }

We have interrupted the flow of the expression, we've had to create a new variable (r), we've had to wrap it all in braces. Not cool!

So, I looked around for something that could do this the functional way. Unfortunately, I couldn't find anything. There is something in Scalaz but it looked complicated and maybe a little too general. So I decided (as I often do) to write my own and call it Spy.

One of the design decisions that I had to make was this: I don't want the Spy to have to be instantiated for every user invocation, or even every class that an invocation appears in. Yet I want each invocation/class to be able to customize the spying behavior somewhat. That's where implicits come (again) to the rescue. But the spy-function (the one that actually does something with a String formed from the expression's value) needs to be found. The natural type of the spy-function is String=>Unit but it turns out that the implicits mechanism couldn't deal with something so ordinary so I changed it to String=>Spy where the Spy class is essentially just a wrapper that has no real significance. Then, the implicit value for the spy-function (if any) could be found and used.

One example of a need to customize the implicit value is to forget about logging and simply write to the console. I tried to make this as easy as possible. See the specification in the repo (linked at the bottom) for an example of this.

Here, using the default slf4j logging mechanism, is the map2 function with logging. Note that, when you use the default mechanism, you must provide an implicit value of a logger in scope. A convenience method has been provided for this as shown below.

  implicit val logger = Spy.getLogger(getClass)
  def map2[T, U](to1: Option[T], to2: => Option[T])(f: (T, T) => U): Option[U] = Spy.spy(s"map2($to1,$to2)",for {t1 <- to1; t2 <- to2} yield f(t1, t2))

We test it using the following specification for map2 (unchanged):

  "map2(Option)" should "succeed" in {
    val one = Some(1)
    val two = Some(2)
    def sum(x: Int, y: Int) = x + y
    map2(one, two)(sum) should matchPattern { case Some(3) => }
    map2(one, None)(sum) should matchPattern { case None => }
  }

And the resulting entries in the log file are:

2016-12-06 22:35:04,123 DEBUG com.phasmid.laScala.fp.FP$  - spy: map2(Some(1),Some(2)): Some(3)
2016-12-06 22:35:04,128 DEBUG com.phasmid.laScala.fp.FP$  - spy: map2(Some(1),None): None

All we had to do, for the default logging behavior, was to ensure there was an implicit value of logger in scope and wrap the expression inside an invocation of Spy.spy. Everything is still purely functional. You can even leave the spy invocation in place if you really want to and either explicitly switch it off by adding false as a third parameter or by turning off debugging in the logger.

If you're interested in using this you can find the Spy class, together with its SpySpec, in my LaScala project.

A generic tail-recursive method for tree traversal

In an earlier article, Transforming iteration into tail recursion, I talked about the general functional-programming technique of replacing iteration by recursion. That method is quite simple and is fairly obvious for linear collections.

But this week, I was working on the general problem of tree traversal. In a binary tree, for instance, as you recurse through the tree, you need to keep track of the results of two recursive tree traversals at each node. They can't both be tail-recursive! But, fortunately, there is a well-known, more general technique available. The method basically consists, as usual, of creating an "inner" method with an accumulator (r for result) and a list of unprocessed elements (ts). The only extra complication is that when we visit a tree node, we need to add all the children (two in the case of a binary tree) to the unprocessed list at the same time as we process the node itself (if there is indeed anything at the node).

Because trees can take so many forms, I created a very general mechanism to implement this "inner" method, which I call traverse in the companion object of a type-class called Parent.

import scala.annotation.tailrec

/**
  * A trait which expresses parenthood.
  * This defines a type class.
  *
  * @tparam T the underlying type of the Parent
  */
trait Parent[T] {
  /**
    * Get the children of a T.
    *
    * @param t the parent whose children we want
    * @return the children as a Seq of T
    */
  def children(t: T): Seq[T]
}

object Parent {
  /**
    * Tail-recursive method to traverse a tree-like object (defined as something that implements Parent).
    *
    * @param f  the map function which takes a T and produces an R for just that parent object (not its children).
    * @param g  the reduce function which accumulates the result (the accumulator, i.e. r, is passed as its first parameter).
    * @param q  the quick return function which, if q(r) yields true, an immediate return of r is made.
    * @param z  the function which builds the T list (ts) from the existing T list and the given T (parent node).
    * @param ts a list of Ts to be worked on.
    * @param r  the current value of the result, i.e. the "accumulator".
    * @tparam T a type which extends Parent, and thus has children of type T -- this "context bound" is implemented via a compiler-generated implicit parameter of type Parent[T].
    * @tparam R the result type.
    * @return a value of R.
    */
  @tailrec
  final def traverse[T: Parent, R](f: T => R, g: (R, R) => R, q: R => Boolean, z: (List[T], T) => List[T])(ts: List[T], r: R): R =
    if (q(r))
      r
    else
      ts match {
        case Nil => r
        case h :: t => traverse(f, g, q, z)(z(t, h), g(r, f(h)))
      }
}

In order to use this for something like size, we need a tree structure made up of nodes, of type T, which implement Parent[T]. In the following code, Node[A] is a node in a tree.

/**
  * This trait expresses the notion of a Node from a tree.
  *
  */
sealed trait Node[+A] {
  /**
    * @return If this node contains an actual value x, return Some(x) else None
    */
  def get: Option[A]

  /**
    * Get the children of this Node, if any.
    * @return the children as a Seq of Nodes
    */
  def children: Seq[Node[A]]

  /**
    * @return the size of the subtree
    */
  def size: Int

  /**
    * @return true if node contains value or children; otherwise false
    */
  def isEmpty: Boolean

  /**
    * @param b the value to be searched
    * @tparam B the type of b
    * @return true if value b is found in the sub-tree defined by this
    */
  def includes[B >: A](b: B): Boolean
}

But, how can we make Node[A] extend Parent[Node[A]]? How about using the magic of type classes (see my earlier blog, Type classes, on the subject)? But a simple type class uses an implicit object which obviously can't itself have a parametric type.

The solution lies in an implicit conversion method. Normally, such methods convert from one type (T1) to another (T2) when the context calls for a T2 and only a T1 is available. However, in this case, T2 is Node[A] but there is no T1! But, the implicit mechanism works also for Unit=>T2. That's really cool!

So, here's the code that takes care of the implicit "evidence" for a node to be a parent:

object Node {
  /**
    * Implicit definition of a Parent for the (type-parameterized, i.e. generic) type Node[A]
    *
    * @tparam A the underlying node type
    * @return a Parent of type Node[A]
    */
  implicit def nodeParent[A]: Parent[Node[A]] = new Parent[Node[A]] {
    /**
      * @return the children as a Seq of Nodes
      */
    def children(t: Node[A]): Seq[Node[A]] = t.children
  }
}

And, finally, here is the code for the implementation of size and includes in a trait TreeLike[+A] which extends Node[A]:

  def size: Int = Parent.traverse[Node[A], Int]({
    case Empty => 0
    case _ => 1
  }, _+_, { x => false}, _ ++ _.children)(List(this), 0)

  def includes[B >: A](b: B): Boolean = Parent.traverse[Node[B], Boolean]({ n => n.get match {
    case Some(`b`) => true
    case _ => false
  }}, _|_, {x => x}, _ ++ _.children)(List(this), false)

Type classes come to the rescue again and we are able to define a wholly generic tail-recursive method for tree traversal.