A Comparison of Distributed Machine Learning Platforms

When doing evaluations there are a lot of nuances.

I will comment on a real world example from our own experience.

Disclaimer: I am biased running a company implementing my own platform. I will try to stick to topics that folks should "look for" when evaluating these things rather than inviting a debate on "my tool vs yours"

For deep learning on spark specifically, there are 2 workloads, training and inference.

Your data pipeline here matters a lot. A job that pulls data from hive will surprise be bottlenecked by hive. This by extension means the kind of data and small things like the number of records you fetch or cache on say: hdfs matters a TON.

On our platform, we implement 2 versions of distributed communications: spark driver and parameter server. The parameter server is faster. The latter is what mxnet and tensorflow do as well. Tensorflow uses grpc underneath. mxnet writes its own.

While both are data flow graphs, the latter is optimized for linear algebra and I would like to say has very different requirements (for example gpus which are another level of complexity)

When using these parameter servers, you should look at what guarantees these things provide. For example, do you need consistent latency? Fault tolerance? At what scale?

Something as simple as"do these parameter servers do p2p on each node?"

With spark, it's not well designed for that. A lot of workers are meant to be stateless.

With spark another gotcha (especially on the linear algebra part) is on heap memory. When you are using say: pyspark for example, how you handle data movement matters a lot, and again surprise ends up being your bottleneck.

We use pyjnius instead in our python interface for linear algebra: https://github.com/deeplearning4j/jumpy

We implemented this because we liked the idea of being able to pass numpy pointers straight to the JVM rather than copying data. I believe pytorch and co do this as well (which are just straight C/C++). That is the right way to do it

The default is py4j (which overall is more versatile but not meant for DL)

Don't even get me started on a normal etl pipeline vs "images", which is typically what you see.

Another big complexity with say: spark and even gpus is gpu usage. When using mesos, you can only reserve a whole gpu, not fractions.

Another thing is worker setup. For deep learning specifically, What BLAS did they use? How was it compiled? Was it optimized for the right cpu architecture?

How does the the workers communicate with the gpu, if any?

If you're using a JVM, how do they handle the FFI?

How are workers configured for cpu/cpu? Things like OMP_NUM_THREADS on each worker matter a lot.

If you're using spark, did you configure k small workers or 1 big one? How did that interact with your multi threaded linear algebra?

Hopefully this gives people an idea of how complicated these things get.

Nevertheless, during my research I found that local work matters a lot! Especially when you have large communication constraints. I addressed this in my master thesis [2], where I introduce 2 novel distributed optimization techniques based on a redefinition of parameter staleness. In this redefinition of parameter staleness, I define staleness in terms of the distance (or difference) between two parameterizations. In a sense, this allows you to automatically tune the amount of work done by a worker. For instance, if multiple workers compute gradients based on older parameterizations which were close to each other, they do not inflict negative work. Imagine it as follows, at the start of an optimization process, you will have relatively large gradients compared to when you are close to an optimum. As a result, having a lot of asynchronous workers at the start of the optimization process (using DOWNPOUR) could really hurt the convergence capabilities, and could even cause divergence (for GIF see [3] and [4]).

To counter this, I introduce AGN (Accumulated Gradient Normalization) as a mechanism to compute a better single gradient update based on increased local exploration. Basically, you compute a sequence of first-order gradients, and then normalize them with the number of "exploration steps" to produce a better parameter server update (this also reduces the magnitude of the vector, reducing the amount of staleness injected in the system, see Chapter 3 of my thesis).

Nevertheless, divergent behavior can still be observed as the number of asynchronous workers increases. This can be countered by constructing a per-weight learning rate based on the difference between the parameterization of the central variable (parameterization held by the parameter server), and the - current - parameterization of the worker (see Chapter 4). Intuitively, this mechanism will nullify the contributions of workers which are "too far" from the current central variable.

[1] http://stanford.edu/~imit/tuneyourmomentum/theory/ (paper in blogpost) [2] https://github.com/JoeriHermans/master-thesis [3] Animation convergence http://joerihermans.com/media/blog/ddl/downpour.gif [4] Animation divergence http://joerihermans.com/media/blog/ddl/downpour_momentum.gif [5] https://github.com/cerndb/dist-keras