The developer can decide to persist and/or cache an RDD for future usage. An RDD may persist in memory only or on disk only—in memory if available, or on disk otherwise as deserialized or serialized Java objects. For instance, an RDD, rdd, can be cached through serialization through a simple statement, as shown in the following code:

rdd.persist(StorageLevel.MEMORY_ONLY_SER).cache

Scala supports lazy values natively. The left-hand side of the assignment, which can either be a value, object reference, or method, is performed once, that is, the first time it is invoked, as shown in the following code:

class Pipeline {
  lazy val x = { println("x"); 1.5}   
  lazy val m = { println("m"); 3}   
  val n = { println("n"); 6}   
  def f = (m <<1)
  def g(j: Int) = Math.pow(x, j)
}
val pipeline = new Pipeline  //1
pipeline.g(pipeline.f)  //2

The order of the variables printed is n, m, and then x. The instantiation of the Pipeline class initializes n but not m or x (line 1). At a later stage, the g method is called, which in turn invokes the f method. The f method initializes the m value it needs, and then g initializes x to compute its power to m <<1 (line 2).

Spark applies the same principle to RDDs by executing the transformation only when an action is performed. In other words, Spark postpones memory allocation, parallelization, and computation until the driver code gets the result through the execution of an action. The cascading effect of invoking all these transformations backward is performed by the direct acyclic graph scheduler.

Spark is implemented in Scala, so you should not be too surprised to know that the most relevant Scala higher methods on collections are supported in Spark. The first table describes the transformation methods using Spark, as well as their counterparts in the Scala standard library. We use the (K, V) notation for (key, value) pairs:

Action methods trigger the collection or the reduction of the datasets from all partitions back to the driver, as listed here:

Scala methods such as fold, find, drop, flatten, min, max, and sum are not currently implemented in Spark. Other Scala methods such as zip have to be used carefully, as there is no guarantee that the order of the two collections in zip is maintained between partitions.

In a perfect world, variables are immutable and local to each partition to avoid race conditions. However, there are circumstances where variables have to be shared without breaking the immutability provided by Spark. To this extent, Spark duplicates shared variables and copies them to each partition of the dataset. Spark supports the following types of shared variables:

The four design principles can be summarized in the following diagram:

An interaction between the Spark driver and RDDs

The preceding diagram illustrates the most common interaction between the Spark driver and its workers, as listed in the following steps:

If you take a look at it closely, the management of datasets and RDDs by the Spark driver is not very different from that by the Akka master and worker actors of futures.

The easiest way to learn Spark is to deploy a localhost in standalone mode. You can either deploy a precompiled version of Spark from the website, or build the JAR files using the simple build tool (sbt) or Maven [12:14] as follows:

Apache supports multiple deployment modes:

The communication between a master node (or driver), cluster manager, and set of slave (or worker) nodes is illustrated in the following diagram:

The communication between a master, slave nodes, and a cluster manager

Use any of the following methods to use the Spark shell:

The output will be as follows:

A partial screenshot of the Spark shell command line output

MLlib is a scalable machine learning library built on top of Spark. As of version 1.0, the library is a work in progress.

The main components of the library are as follows:

The machine learning bytecode is conveniently included in the Spark assembly JAR file built with the simple build tool.

The transformation and actions are performed on RDDs. Therefore, the first step is to create a mechanism to facilitate the generation of RDDs from a time series. Let's create an RDDSource singleton with a convert method that transforms a time series xt into an RDD, as shown here:

def convert(
    xt: immutable.Vector[DblArray], 
    rddConfig: RDDConfig) 
    (implicit sc: SparkContext): RDD[Vector] = {

  val rdd: RDD[Vector] = 
     sc.parallelize(xt.toVector.map(new DenseVector(_))) //3
  rdd.persist(rddConfig.persist) //4
  if( rddConfig.cache) rdd.cache  //5
  rdd
}

The last rddConfig argument of the convert method specifies the configuration for the RDD. In this example, the configuration of the RDD consists of enabling/disabling cache and selecting the persistency model, as follows:

case class RDDConfig(val cache: Boolean, 
    val persist: StorageLevel)

It is fair to assume that SparkContext has already been implicitly defined in a manner quite similar to ActorSystem in the Akka framework.

The generation of the RDD is performed in the following steps:

Once the RDD is created, it can be used as an input for any algorithm defined as a sequence of transformation and actions. Let's experiment with the implementation of the K-means algorithm in Spark/MLlib.

The first step is to create a SparkKMeansConfig class to define the configuration of the Apache Spark K-means algorithm, as follows:

class SparkKMeansConfig(K: Int, maxIters: Int, 
     numRuns: Int = 1) {   
  val kmeans: KMeans = {      
    (new KMeans).setK(K) //6
      .setMaxIterations(maxIters)  //7
      .setRuns(numRuns) //8
  }
}

The minimum set of initialization parameters for MLlib K-means algorithm is as follows:

The SparkKMeans class wraps the Spark KMeans into a data transformation of the ITransform type, as described in the Monadic data transformation section in Chapter 2, Hello World! The class follows the design template for a classifier, as explained in the Design template for immutable classifiers section in the Appendix A, Basic Concepts:

class SparkKMeans(    //9
    kMeansConfig: SparkKMeansConfig, 
    rddConfig: RDDConfig, 
    xt: Vector[DblArray])
   (implicit sc: SparkContext) extends ITransform[DblArray](xt){

  type V = Int   //10
  val model: Option[KMeansModel] = train  //11

  override def |> : PartialFunction[DblArray, Try[V]] //12
  def train: Option[KMeansModel] 
}

The constructor takes three arguments: the Apache Spark KMeans configuration kMeansConfig, the RDD configuration rddConfig, and the xt input time series for clustering (line 9). The return type of the ITransform trait's partial function |> is defined as an Int (line 10).

The generation of model merely consists of converting the time series xt into an RDD using rddConfig and invoking MLlib KMeans.run (line 11). Once it is created, the model of clusters (KMeansModel) is available for predicting a new observation, x, (line 12), as follows:

override def |> : PartialFunction[DblArray, Try[V]] = {
  case x: DblArray if(x.length > 0 && model != None) => 
     Try[V](model.get.predict(new DenseVector(x)))
}

The |> prediction method returns the index of the cluster of observations.

Finally, let's write a simple client program to exercise the SparkKMeans model using the volatility of the price of a stock and its daily trading volume. The objective is to extract clusters with features (volatility and volume), each cluster representing a specific behavior of the stock:

val K = 8
val RUNS = 16
val MAXITERS = 200
val PATH = "resources/data/chap12/CSCO.csv"
val CACHE = true

val sparkConf = new SparkConf().setMaster("local[8]")
   .setAppName("SparkKMeans")
   .set("spark.executor.memory", "2048m") //13
implicit val sc = new SparkContext(sparkConf) //14

extract.map { case (vty,vol)  => {  //15
  val vtyVol = zipToSeries(vty, vol)  
  val conf = SparkKMeansConfig(K,MAXITERS,RUNS) //16
  val rddConf = RDDConfig(CACHE, 
                    StorageLevel.MEMORY_ONLY) //17

  val pfnSparkKMeans = SparkKMeans(conf,rddConf,vtyVol) |> //18
  val obs = Array[Double](0.23, 0.67)
  val clusterId = pfnSparkKMeans(obs)
}

The first step is to define the minimum configuration for the sc context (line 13) and initialize it (line 14). The vty and vol volatility variables are used as features for K-means and extracted from a CSV file (line 15):

def extract: Option[(DblVector, DblVector)] = {
  val extractors = List[Array[String] => Double](
    YahooFinancials.volatility, YahooFinancials.volume 
  )
  val pfnSrc = DataSource(PATH, true) |>
  pfnSrc( extractors ) match {
    case Success(x) => Some((x(0).toVector, x(1).toVector))
    case Failure(e) => { error(e.toString); None }
  }
}

The execution creates a configuration config for the K-means (line 16) and another configuration for the Spark RDD, rddConfig, (line 17). The pfnSparkKMeans partial function, which implements the K-means algorithm, is created with the K-means, RDD configurations, and the input data vtyVol (line 18).

The performance of a Spark application depends greatly on the configuration parameters. Selecting the appropriate value for those configuration parameters in Spark can be overwhelming—there are 54 configuration parameters as of the last count. Fortunately, the majority of those parameters have relevant default values. However, there are few parameters that deserve your attention, including the following:

The purpose of the test is to evaluate how the execution time is related to the size of the training set. The test executes K-means from the MLlib library on the volatility and trading session volume on the Bank of America (BAC) stock over the following periods: 3 months, 6 months, 12 months, 24 months, 48 months, 60 months, 72 months, 96 months, and 120 months.

The following configuration is used to perform the training of K-means: 10 clusters, 30 maximum iterations, and 3 runs. The test is run on a single host with 8-CPU cores and 32 GB of RAM. The test was conducted with the following values of parameters:

The first step after executing a test for a specific dataset is to log in to the Spark monitoring console at http://host_name:4040/stages:

The average duration of the K-means clustering versus size of trading data in months

Obviously, each environment produces somewhat different performance results but confirms that the time complexity of the Spark K-means is a linear function of the training set.

This test barely scratches the surface of the capabilities of Apache Spark. The following are the lessons learned from personal experience in order to avoid the most common performance pitfalls when deploying Spark 1.3+: