Monday, July 9, 2018

Speed up Fibonacci Sequence with Tail Recursion

Target audience: Intermediate
Estimated reading time: 5'

The Fibonacci sequence is characterized by each number being the summation of its two immediate predecessors. In this article, we delve into an examination and comparison of tail recursion's efficiency in Scala, specifically when applied to the Fibonacci sequence.


Table of contents
Follow me on LinkedIn

There are many ways to skin a cat and implement the Fibonacci recursion. This post illustrates the power of tail recursion in Scala, relative to alternative solution such as recursion without tail elimination and iterations. This post compares the performance of the simple "out-of-the-box" implementation of the Fibonacci recursion with an Scala implementation using tail-recursion.

Note: This post relies on Scala 2.11.8


Non-tail recursive Fibonacci

The Fibonacci algorithm is defined as
   f(n) = f(n-1) + f(n-2)
   f(0) = 0
   f(1) = 1
Let's look at text book recursive implementation.

 
def fibonacci(n: Int): Long =  {
    if(n == 0) 0
    else if(n == 1) 1
    else fibonacci1(n-1) + fibonacci1(n-2)
}

The duration of execution is recorded for values between 2 and 50, run 10,000 times.
The profile of the performance of the non-tail recursion of Fibonacci should not be a surprise. After all, the time complexity of this implementation is a staggering O(2n)


Tail recursion

Let's look at one example of implementation of the Fibonacci formula using the Scala tail recursion.

def fibonacci(n: Int): Long = {
   
   @tailrec
   def _fibonacci(i: Int, next: Long, sum: Long): Long = 
       if(i ==0) sum
       else fibonacci(i-1, sum+ next, next)

   fibonacci(n, 1, 0)
}

The performance of the tail recursion is measure against a plain-vanilla iterative execution. The duration of execution is recorded for values between 5 and 200, run 100,000 times.

 
def fibonacci(n: Int): Long = {
    var s1 = 0
    var s2 = 1 
 
    (n-1 to 0 by -1).foreach(
       _ => {
           val tmp = s2 + s1
           s1 = s2
           s2 = tmp    
      }
   )
   
   s1
}

The following chart illustrates that the performance of the implementation of Fibonacci using the tail recursion and the iterative execution is almost identical.

Iterative computation

For good measure, I thought it would be interesting to measure the performance of an iterative approach using a fold.

def fibonacci(n: Int): Long = 
 
  (n-1 to 0 by -1)./:((0, 1)) {  
      case ((s1, s2), n) => { 
          val tmp = s2 + s1
          (s2, tmp) 
      }
   }._1

Here is the performance profile for the iterative computation.
Note: The fold (:/) can be implemented as a closure using a recursive formula for the function argument.

Conclusion

Scala tail recursion is indeed as effective as the iterative implementation at processing the Fibonacci formula. The performance of the recursion without tail elimination is extremely poor. Aggregators or reducers such as fold, reduce and scan add some overhead to the computation although the performance is still decent.

Thank you for reading this article. For more information ...

References



---------------------------
Patrick Nicolas has over 25 years of experience in software and data engineering, architecture design and end-to-end deployment and support with extensive knowledge in machine learning. 
He has been director of data engineering at Aideo Technologies since 2017 and he is the author of "Scala for Machine Learning" Packt Publishing ISBN 978-1-78712-238-3

Thursday, May 24, 2018

Streaming Moving Averages with Apache Spark

Target audience: Intermediate
Estimated reading time: 5'

Spark Streaming offers an abstraction called Discretized Stream, or DStream, which signifies a continuous flow of data. This can either stem from an initial source or emerge as a result of transforming the input data. Essentially, a DStream is characterized by a consistent sequence of Resilient Distributed Datasets (RDD) [1].


Follow me on LinkedIn
Important notes
  • This post describes the streaming features as implemented in Apache Spark 2.0.x 
  • Stream should not confused with structured streaming [2].

Use case

The creation of models through supervised learning from a given training sets usually requires the application of some pre-processing algorithm such as smoothing, filtering or extrapolating existing data for missing values. 

Computing the moving average of a time series is a simple and convenient technique used to filter out noise and reduce the impact of outliers on trend lines. For this post, we consider the simplest form of moving average over a period of p values, as defined in the following formula \[\tilde{x_{t}}= \frac{1}{p}\sum_{j=t-p+1}^{t}x_{j}\] 
The iterative (or recursive) form is defined as \[\tilde{x_{t}}=\tilde{x_{t-1}}+\frac{1}{p}(x_{t}-x_{t-p})\] 

Spark DStreams

This section refers to discretized streams that were introduced in Spark 1.x. Structured streaming will be discussed in a future post.



Thank you for reading this article. For more information ...

References



---------------------------
Patrick Nicolas has over 25 years of experience in software and data engineering, architecture design and end-to-end deployment and support with extensive knowledge in machine learning. 
He has been director of data engineering at Aideo Technologies since 2017 and he is the author of "Scala for Machine Learning" Packt Publishing ISBN 978-1-78712-238-3

Monday, April 23, 2018

Overload Operators in C++ & Scala

Target audience: Beginner
Estimated reading time: 4'

Operator overloading allows for a reinterpretation of existing operators by imparting a distinct significance to them, all while preserving their original function. This is an example of compile-time polymorphism
While C++ offers the capability to assign unique meanings to operators for specific data types, how does its approach align with that of Scala's operator overloading?


Table of contents
Follow me on LinkedIn
 

Overview

In object-oriented programming, operator overloading is a specific case of polymorphism, where different operators may have different implementations depending on their arguments. The main benefit of operator overloading is to allow the developer to program using notation relevant to the domain (i.e. summation of vectors) and allows user-defined types a similar level.

C++ and Scala supports overloaded or redefinition of operators.
There is however a distinction between C++ and Scala in terms of operator overloading: In Scala the operators are over-loadable by the programmer who create methods with operator name while in C++ the operators are over-loadable by the language as a predefined set.

Notes
  • For the sake of readability of the implementation of algorithms, all non-essential code such as error checking, comments, exception, validation of class and method arguments, scoping qualifiers or import is omitted.
  • Scala version used in 2.11.8

C++ operator overloading

In C++, operator overloading instructs the compiler to perform a certain operation when its corresponding operator is used on one or more variables. However, the && , || and comma operators cannot be overloaded.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
class Point {
private:
   int x;
   int y;
public:
      // Constructor
   Point(int x = 0, int y = 0) : x(x), y(y) { }

      // overloading += operator
   Point& operator+=(const Point& p) { 
       this.x += x;
       this.y += y;
       return (*this);
    }
  
   Point operator+(const Point& p) {
       Point result;
       result.x = x + p.getX();
       result.y  = y + p.getY();
       return result, 
   }

   friend std::ostream& operator<<(std::ostream& out, Point p) {
        out << "(" << p.getX() << "," << p.getY() << ")";
        return out;
   }
};


int main() {
    Point p1(4, 5), p2(5, 11)
    cout << "p1=" << p1 << "p1 + p2=" << p1+p2;
    return 0;
}

Scala overloading by proxy

Scala does not technically have operator overloading because it does not have operators in the common sense.  The operators are indeed function name defined by the programmer. Scala treats all operators as methods and thus allows operator overloading by proxy.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
final class Point(val x: Int, val y: Int) {
 
   // Binary operator
   def +(p : Point) : Point = {
       require(p != null, "cannot add a null point")
       new Point(x + p.x, y + p.y)
    }
    
    override def toString : String = s"(${x.toString},${y.toString)"           
}
 
object Overload extends App {
    val p1 = new Point(2, 0)
    val p2 = new Point(4, -2)
       
    Console.println(s"p1=$p1 and p1+p2=${(p1 + p2)}")
}

The class Point is immutable as the coordinate, x and y are declared private value. It is highly recommended to create immutable objects and containers because it reduced the object state lifecycle. Immutability avoids the need of complex and error-prone concurrent update of the state of an object (race condition).

Thank you for reading this article. For more information ...

References

  • Scala for the Impatient - C. Horstman - Addison-Wesley 2012
  • The Annotated C++ Reference Manual - M. Ellis,  B. Stroustrup - Addison-Wesley 1991
  • github.com/patnicolas

---------------------------
Patrick Nicolas has over 25 years of experience in software and data engineering, architecture design and end-to-end deployment and support with extensive knowledge in machine learning. 
He has been director of data engineering at Aideo Technologies since 2017 and he is the author of "Scala for Machine Learning" Packt Publishing ISBN 978-1-78712-238-3

Friday, March 16, 2018

Fisher-Yates Shuffle in Scala

Target audience: Intermediate
Estimated reading time: 4'


This post describes the Fisher-Yates algorithm to shuffle randomly a sequence of objects.
Follow me on LinkedIn

Note: The code associated with this article is written using Scala 2.12.4

Overview

The generation of a sequence of randomly ordered elements from an existing sequence is a common problem. How can you be sure if the new sequence of items is actually randomly generated?
The Fisher-Yates shuffle algorithm (a.k.a. Donald Knuth shuffle) produces unbiased permutations with a similar likelihood.
One important benefit of Fisher-Yates is the ability to shuffle the elements of the sequence, in place.


Note:
There are several interpretations of the shuffling algorithm. This post describes the implementation of the most common Fisher-Yates algorithm


Fisher-Yates

Here is the Fisher-Yates in a nutshell. Let's consider an array arr of n elements
For each array index k from n-1 to 1
    Select a random value m  between k and 0
    Swap arr[r] and arr[k]
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
def fisherYates[T](t: Seq[T]): Seq[T] = {

   setSeed(System.currentTimeMillis)
  
   (0 until t.size).foreach (n => {
       val randomIdx = n + nextInt(t.size-n)
       val tmp = t(randomIdx)
       t.update(randomIdx, t(n))
       t(n) = tmp
   })
   t
} 

The Fisher-Yates algorithm executes as an iteration with the following two steps:
  • Select a random item in the remaining sequence (line 6)
  • Swap the randomly selected item with the current item (lines 8,9)

Fast Fisher-Yates implementation

The performance of the Fisher-Yates algorithm can be improved by implementing the swapping of the randomly selected item with the current time in-place using bit manipulation operators.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
def fisherYates2(seq: mutable.Seq[Int]): IndexedSeq[Int] = {
  require(seq.size > 0, "Undefined argument")

  setSeed(System.currentTimeMillis)
  (0 until seq.size).map(i => {
     var randomIdx: Int = i + nextInt(seq.size-i)
     seq(randomIdx) ^= seq(i)
     seq(i) = seq(randomIdx) ^ seq(i) 
     seq(randomIdx) ^= (seq(i))
     seq(i)
  })
}

The following code snippet tests the two implementations of the Fisher-Yates shuffling algorithm.
  val input = Array[Char](
    'a', 't', '4', '&', '9', '2', 'z', 'y', 'r', '8', '*', '?'
  )
  println(fisherYates[Char](input).mkString(","))
  // t,&,*,2,a,r,4,z,?,y,8,9

  println(fisherYates2(Array.tabulate(34)(n => n)).mkString(","))
  // 9,24,11,29,1,31,28,20,6,18,23,0,13,17,22,0,27,
  // 4,16,7,25,14,30,32,5,12,8,19,21,0,0,33,26,0
Note: Scala standard library provides developers with a shuffling method: Random.shuffle.

Thank you for reading this article. For more information ...

References


---------------------------
Patrick Nicolas has over 25 years of experience in software and data engineering, architecture design and end-to-end deployment and support with extensive knowledge in machine learning. 
He has been director of data engineering at Aideo Technologies since 2017 and he is the author of "Scala for Machine Learning" Packt Publishing ISBN 978-1-78712-238-3

Wednesday, February 7, 2018

Implementing AdaGrad Optimizer in Spark

Target audience: Intermediate
Estimated reading time: 5'

Have you ever desired a Gradient Descent approach with greater flexibility? If that's the case, Adaptive Gradient Descent (AdaGrad) could be a suitable alternative.


What you will learn: The workings of the Adaptive Gradient algorithm, how it improves upon Stochastic Gradient Descent, and its implementation using Apache Spark.



Table of contents
Follow me on LinkedIn
Notes
  • This post assumes that reader has rudimentary knowledge of the Scala API of Apache Spark and basic understanding of machine learning.
  • The code associated with this article is written using Scala 2.12.11 and Spark 3.1.0

Background

The optimization algorithm known as Stochastic Gradient Descent (SGD) is frequently employed to minimize the loss function during the training of various machine learning models, including support vector machines, logistic regression, and back-propagation neural networks. Typically, in its most basic form, SGD calculates the gradient using a singular learning rate.

It's quite usual to find that the features of a model exhibit significant variance across different observations. Under such circumstances, an adaptive gradient (AdaGrad) algorithm that assigns a unique learning rate to each feature could be an effective solution. There exist numerous methods to develop an algorithm that assigns a distinct learning rate to every feature. 
Accidentally, Spark ML library, MLlib, does not include an implementation of AdaGrad.
This article discusses the AdaGrad algorithm and an implementation for Apache Spark.

Stochastic Gradient Descent (SGD)

Apache Spark stands out as a rapid and versatile cluster computing system, offering advanced APIs in Java, Scala, Python, and R, along with an enhanced engine capable of supporting general execution graphs.

Within the Apache Spark ecosystem lies MLlib, a machine learning library. The stochastic gradient descent optimizer within this library serves as a randomized variant of the (batched) gradient descent algorithm, which is employed for minimizing a continuous differentiable objective function. In the realm of supervised machine learning, this objective function typically manifests as a loss function, such as logistic loss, sum of least squares, and others. \[L(w)=\frac{1}{n}\sum_{I=1}^{n}(y_{i}-f(x_{i}|w))^{2}\] The objective function L is expressed as the summation of differentiable functions. In supervised learning, the loss related to a specific feature is defined as a continuous, differentiable, convex function. \[L(w)=\sum_{i=1}^{n}L_{i}(w)\] In supervised learning, the vector w represents the vector of weights (or model parameters). At each iteration of the stochastic gradient descent, the weights are updated using the formula \[w_{t+1}=w_{t}-\eta \sum_{i=0}^{n}\frac{\partial L}{\partial w_{i, t}}\] Stochastic Gradient Descent (SGD) aims to reduce the loss function, which represents the difference between the model's predicted values and the expected values. In each iteration, SGD picks a small group of observations, referred to as a mini-batch, for the model's training. This repetitive procedure is designed to progressively approach the true global minimum of the loss function.

Adaptive Gradient Descent

The core concept of AdaGrad revolves around the strategy of boosting the learning rate for sparse features (or model parameters) while reducing it for more dense features. As a result, AdaGrad enhances the efficiency of converging towards the minimum loss in models with sparse features, particularly when these sparse features hold significant information. \[w_{t+1}=w_{t} -\frac{1}{\sqrt{\sum_{t=1}^{T}\bigtriangledown _{ti}^{t} + \varepsilon }}\frac{\partial L}{\partial w_{ti}}\]

SGD in Apache Spark

The Apache spark MLlib library has two implementations of SGD
  • Generic Gradient Descent and related classes in the mllib.optimization package
  • SGD bundled with classifier or regression algorithms such as LogisticRegressionWithSGD, LassoWithSGD, SVMWithSGD or RidgeRegressionWithSGD

We plan to utilize the optimization package to modify the stochastic gradient descent. Our goal is to use the mllib.optimization.GradientDescent template class from MLlib and create a custom Updater that implements an adaptive gradient with per-feature learning rates.

This Updater is responsible for "updating the model's weights" (in models like Logistic Regression or SVM) by multiplying the current learning rate with the partial derivative of the loss with respect to each weight, as previously described. We'll name our custom Updater as AdaGradUpdater, which will handle the model weight updates using the adaptive gradient approach. Following this, we'll initialize the SGD in the following manner.
   val adaSGD = new GradientDescent.
                    .setUpdater(new AdaGradUpdater)
                    .setStepSize(0.01)
                    . .....
The class AdaGradUpdater has to implement the generic compute method
  Updater.compute(
      oldWeights: Vector, 
      gradient: Vector, 
      stepSize: Double, 
      iter: Int, 
      regCoefs: Double
  ): (Vector, Double)
The method returns the tuple (vector of new weights, loss). Let's implement the AdaGrad algorithm


Implementation of AdaGrad

As mentioned earlier, the implementation of AdaGrad consists of overriding the method Updater.compute.

The computation of the learning rate requires us to record the past values of the square value of the gradient (previous steps) for this particular weight, in the array gradientHistory (line 2). First we define the method += to update the gradient history (lines 26-367. The first call to the method creates the gradient history (line 30).
The next step consists of converting the existing (old) weights into a Breeze dense vector brzWeights (line 13). The array of the new learning rates is computed as the inverseVelocity coefficient (line 40).

The learning rates are zipped with the old weights (line 14) to update the weights newWeights as a new dense vector(line 15-17). The linear algebra (matrix computation) on the Spark data node is actually performed by the LINPACK library under the cover through calls to brzAxpy (line 20) and brzNorm (line 21).

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
 final class AdaGradL2Updater(dimension: Int) extends Updater {
  var gradientsHistory: Array[Double] = _

  override def compute(
       weightsOld: Vector,
       gradient: Vector,
       stepSize: Double,
       iter: Int,
       regParam: Double
  ): (Vector, Double) = {

    +=(gradient)
    val brzWeights: BV[Double] = weightsOld.toBreeze.toDenseVector
    val sumSquareDerivative = inverseVelocity.zip(brzWeights.toArray)
    val newWeights: BV[Double] = 
       new DenseVector[Double](sumSquareDerivative.view.map {
          case (coef, weight) => weight * (1.0 -regParam * coef)
       }

    brzAxpy(-1.0, gradient.toBreeze, newWeights)
    val norm = brzNorm(brzWeights, 2.0)

    (Vectors.fromBreeze(brzWeights), 0.5 * regParam * norm * norm)
  }


  private def +=(gradient: Vector): Unit = {
    val grad = gradient.toArray
    
    grad.view.zipWithIndex.foreach {
      case (g, index) => {
        if(gradientsHistory == null)
          cgradientsHistory = Array.fill(grad.length)(0.0)

        val existingGradient = gradientsHistory(index)
        gradientsHistory.update(index, existingGradient + g*g)
      }
    }
  }


  def inverseVelocity = gradientsHistory.map(1.0/Math.sqrt(_))
}

Analysis

Given a logistic regression model to classifier of time series pattern, we compute and compare the training binary cross-entropy loss without regularization using the default MLlib SGD and our implementation of AdaGrad.

The AdaGrad algorithm has its limitation. The accumulation can grow very large over time, causing the learning rate to shrink and become infinitesimally small, which effectively stops the parameter from updating.

AdaGrad is not the only alternative to traditional SGD. RMS and Adam are worth investigating. 

Thank you for reading this article. For more information ...

References


---------------------------
Patrick Nicolas has over 25 years of experience in software and data engineering, architecture design and end-to-end deployment and support with extensive knowledge in machine learning. 
He has been director of data engineering at Aideo Technologies since 2017 and he is the author of "Scala for Machine Learning" Packt Publishing ISBN 978-1-78712-238-3