# TensAIR: Online Learning from Data Streams via Asynchronous Iterative
Routing

Mauro Dalle Lucca Tosi  Department of Computer Science
University of Luxembourg
Esch-sur-Alzette, Luxembourg
0000-0002-0218-2413  Vinu E. Venugopal  ScaDS Lab
IIIT Bangalore
Karnataka, India
0000-0003-4429-9932  Martin Theobald  Department of Computer Science
University of Luxembourg
Esch-sur-Alzette, Luxembourg
0000-0003-4067-7609

###### Abstract

Online learning (OL) from data streams is an emerging area of research which
encompasses numerous challenges from stream processing, machine learning, and
networking. Recent extensions of stream-processing platforms, such as Apache
Kafka and Flink, already provide basic extensions for the training of neural
networks in a stream-processing pipeline. However, these extensions are not
scalable and flexible enough for many real-world use-cases, since they do not
integrate the neural-network libraries as a first-class citizen into their
architectures.

In this paper, we present TensAIR, which provides an end-to-end dataflow
engine for OL from data streams via a protocol to which we refer as
asynchronous iterative routing. TensAIR supports the common dataflow
operators, such as Map, Reduce, Join, and has been augmented by the data-
parallel OL functions train and predict. These belong to the new Model
operator, in which an initial TensorFlow model (either freshly initialized or
pre-trained) is replicated among multiple decentralized worker nodes. Our
decentralized architecture allows TensAIR to efficiently shard incoming data
batches across the distributed model replicas, which in turn trigger the model
updates via a form of asynchronous stochastic gradient descent. We empirically
demonstrate that TensAIR achieves a nearly linear scale-out in terms of (1)
the number of worker nodes deployed in the network, and (2) the throughput at
which the data batches arrive at the dataflow operators. We exemplify the
versatility of TensAIR by investigating both sparse (Word2Vec) and dense
(CIFAR-10) use-cases, for which we are able to demonstrate very significant
performance improvements in comparison to Kafka, Flink, and the Horovod
distributed learning framework recently developed by Uber. We also demonstrate
the magnitude of these improvements by depicting the possibility of real-time
concept drift adaptation of a sentiment analysis model trained over a Twitter
stream.

###### Index Terms:

Online Learning, Neural Networks, Asynchronous Stream Processing, Asynchronous
Stochastic Gradient Descent

##  I Introduction

Machine learning (ML) has become ubiquitous in modern data-analytics and
decision-making tasks. Current ML solutions demand a considerable amount of
training time, which usually renders it infeasible to retrain the entire ML
machinery “from scratch” whenever new data arrives. Online learning (OL) is a
branch of ML that studies solutions to time-sensitive problems which demand
real-time answers based on fractions of data received in the form of data
streams [1]. Moreover, a common characteristic of data streams is the presence
of concept drifts [2], i.e., changes in the statistical properties among the
incoming data objects over time. To adapt to concept drifts, one may rely on
either passive or active adaptation strategies [3]. The passive strategy
updates the trained model indefinitely, with no regard to the actual presence
of concept drifts. Active drift adaptation strategies, on the other hand, only
adapt the model when a concept drift has been explicitly identified. In OL, a
passive strategy (which is the default strategy we consider in this work) can
be seen as the computationally more demanding approach, since the buffering
capabilities of a stream-processing engine are limited, such that the model
needs to be constantly kept up-to-date with the incoming data batches.

Due to the intrinsic characteristics of OL, especially with respect to real-
world applications, it is not feasible to depend on solutions that spend an
undue amount of time on retraining, nor on those that cannot adapt to concept
drifts. Therefore, until recently, complex OL problems could not rely on
robust solutions common to other ML problems, like those involving Artificial
Neural Networks (ANNs). Training in OL and ANNs is remarkably similar and yet
is not exactly the same. Both are trained on incremental steps but differ with
respect to the source and distribution of the training samples. ANNs sample
their training data repeatedly (using so-called “epochs”) from a pre-defined
dataset with a fixed data distribution [4]. OL approaches, on the other hand,
inherently need to sample data from an unbounded data stream with changing
data distribution [5]. However, by analyzing subsets of a data stream between
two significant concept drifts, in which the data is fixed and its
distribution does not change significantly, both OL and ANN training become
analogous again. Therefore, by overcoming limitations related to the
convergence-time it takes to train ANNs, their usage to solve complex OL
problems could substantially improve current solutions. Despite their apparent
similarities, the usage of ANNs on OL scenarios is not straightforward.
Therefore, we highlight the following challenges:

  * (1)

Multiple epochs are usually necessary to train ANNs, in particular when
modelling complex problems [4]. Meanwhile, OL only allows for processing
incoming data batches once (or at most a limited amount of times), depending
on the general buffering capabilities of the underlying streaming engine.

  * (2)

The convergence-time for training ANNs still is too high for many OL use-
cases. Current systems still rely on libraries tailored to perform offline
learning. Thus, their scale-out performance is limited when training from a
data stream (see, e.g., Flink’s dl-on-flink [6] extension, and Horovod [7]
developed by Uber).

  * (3)

Finally, ANNs are highly sensitive to the availability and quality of labeled
input data, which is easier to be provided and assessed on a fixed dataset
than on a data stream.

Current stream-processing platforms, such as Apache Flink [8] and Kafka [9]
already provide basic extensions [6, 10] for training ANNs under a streaming
setting. However, these extensions are not prepared to sustain real-time
training on a high-throughput scenario due to the following limitations: (1)
Kafka does not currently support any form of distributed training in
TensorFlow [9]; (2) Flink indeed exploits the distributed TensorFlow APIs
(tf.distribute) which, to this end, is however limited to buffering the entire
training data and thereby suffers performance issues under an OL setting (see
Section IV); (3) overall, both of the Kafka and Flink extensions are still
largely tailored to a few pre-defined use-cases and do not generalize well to
the requirements of OL under different data formats, neural-network
configurations, and streaming setups.

Solutions to increase the training efficiency of ANN algorithms, such as
Apache SystemML [11], Horovod [7] and the distributed Tensorflow [12] APIs, on
the other hand, do not natively target OL problems and their characteristics.
Therefore, it is still an open research problem to develop a system that can,
in real-time, train an ANN from a high-throughput data stream. Moreover,
current research on OL is mostly focused on how to improve the quality of the
input data and how to adapt to concept drifts [13, 14, 15, 16, 17]. Typical
scale-up approaches for training ANNs are limited by the computational power
of a single GPU and the server’s memory/CPU capacity. Thus, scale-out
approaches, like, e.g., Horovod [7] used by Uber for several years, come with
their own distributed libraries for training ANNs but are still limited by
their synchronous networking protocols, SGD updates, and further buffering
requirements for the training data.

Finally, under an asynchronous SGD (ASGD) setting [18], workers are allowed to
compute their gradient computations also on stale model parameters. This
behaviour helps to minimize idle times but makes it harder to mathematically
prove SGD convergence. Recent developments on ASGD [19, 20, 21, 22, 23, 24,
25, 26, 27], however, have tackled exactly this issue under different
assumptions. In particular, Koloskova et al. [19] recently proved ASGD
convergence with varying computation and communication frequencies among the
worker nodes within very good error bounds. Moreover, they also proved that
ASGD is always faster than mini-batch SGD—independently of the delay of the
model updates.

Contributions.  In this paper, we present the architecture of TensAIR, an end-
to-end dataflow engine combining our AIR stream-processing engine [28] with
TensorFlow [12], thus yielding a fully distributed OL framework implemented in
C++. Different from current approaches, TensAIR focuses on scaling-out in
particular the training of ANNs via an asynchronous and decentralized
architecture. This architecture relies on asynchronous message passing in
combination with asynchronous SGD updates, thereby eliminating the bottleneck
of a centralized parameter server for training. Its current programming
interface is flexible enough to let users specify their desired TensorFlow
model structure and optimizer (via the Keras Python APIs), as well as the
concept-drift identification algorithm. Although the asynchronous and
decentralized training of ANNs is not a completely new direction [26], we
believe that TensAIR is the first architecture to truly enable the
asynchronous and decentralized training of ANNs under an OL scenario by
incorporating the neural-network components as first-class dataflow
operators—both for training and prediction.

##  II Background & Related Work

###  II-A Online Learning

Online learning (OL) recently gained visibility due to the increase in the
velocity and volume of available data sources compared to the past decade
[29]. OL algorithms are trained using data streams as input, which differs
from traditional ML algorithms that have a pre-defined training dataset.

Streams & Batches.  Formally, a data stream $\mathcal{S}$ consists of ordered
events $e$ with timestamps $s$, i.e.,
$(e_{1},s_{1}),\ldots,(e_{\infty},s_{\infty})$, where the $s_{i}$ denote the
processing time at which the corresponding events $e_{i}$ are ingested into
the system. These events are usually analysed in batches $B_{j}$ of fixed size
$b$, as follows:

$\displaystyle B_{1}$ $\displaystyle=(e_{1},s_{1}),\ldots,(e_{b},s_{b})$
$\displaystyle B_{2}$ $\displaystyle=(e_{b+1},s_{b+1}),\ldots,(e_{2b},s_{2b})$
$\displaystyle\ldots$

Batches $B_{j}$ are analyzed individually. Thus, if processed in an
asynchronous stream-processing scenario, the batches (and in particular the
included events $e_{i}$) can become out-of-order as they are handled within
the system, even if the initial $s_{i}$ were ordered. In common stream-
processing architectures, such as Apache Flink [8], Spark [30] and Samza [31],
batches are distinguished into sliding windows, tumbling windows and (per-
user) sessions [32].

Latency vs. Throughput.  When analyzing systems that process data streams, one
typically benchmarks them by their latency and throughput [33]. Formally,
latency is the time it takes for a system to process an event, from the moment
it is ingested to the moment it is used to produce a desired output.
Throughput, on the other hand, is the number of events that a system can
receive and process per time unit. The sustainable throughput is the maximum
throughput at which a system can handle a stream over a sustained period of
time (i.e., without exhibiting a sudden surge in latency, then called
“backpressure” [34], or even a crash).

###  II-B Artificial Neural Networks

ANNs denote a family of supervised ML algorithms which are designed to be
trained on a pre-defined dataset [35]. A training dataset is composed of
multiple $(x,y)$ pairs, in which $x$ is a training example and $y$ is its
corresponding label. ANNs are usually trained using mini-batches $X$, which
are sets of $(x,y)$ pairs of fixed size $N$ that are iteratively (randomly)
sampled from the training dataset, thus
$X={(x_{i},y_{i}),\ldots,(x_{i+N},y_{i+N})}$.

BGD vs. SGD.  An ANN model is represented by the weights and biases of the
network, described together by $\theta$. We train the network using Batch
Gradient Descent (BGD) [36] which is based on Stochastic Gradient Descent
(SGD) [37]. BGD updates $\theta$ by considering $\nabla L(X,\theta)$, which is
the gradient of a pre-defined loss function $L$ with respect to $\theta$ when
taking $X$ as input. Thus, we can represent the update rule of $\theta$ as in
Equation 1, in which $t$ is the iteration in BGD, and $\alpha$ is a pre-
defined learning rate.

$\theta_{t+1}=\theta_{t}-\alpha\nabla L(X,\theta_{t})$ (1)

Based on Equation 1, $\theta_{t+1}$ is defined based on two terms. The second
term is the more computationally expensive one to calculate, which we refer to
as gradient calculation (GC). The remainder of the equation we call gradient
application (GA), which consists of the subtraction between the two terms and
the assignment of the result to $\theta_{t+1}$.

Convergence.  Despite the differences between OL and ANNs, they have a similar
training pipeline. Both approaches are trained based on analogous sets of data
points, which are the batches for OL and the mini-batches for ANNs. A seminal
result for training ANNs [35] is that SGD converges to a (local) minimum of
the loss function as the number of batches approaches infinity—even when not
all of the actually available training examples fall into these batches. So-
called epochs were introduced to ANN training in order to compensate for the
lack of labeled training examples by performing repeated iterations over the
available examples. A main conjecture we follow with TensAIR thus is that,
instead of using epochs, we may train the underlying ANN by using new batches
from a data stream as input and hence converge as long as the principal
distribution of the $(x,y)$ pairs in the data stream remains unchanged (e.g.,
between two major concept drifts).

###  II-C Distributed Artificial Neural Networks

Over the last years, ANN models have grown in size and complexity.
Consequently, the usage of traditional centralized architectures has become
unfeasible when training complex models due to the high amount of time they
spend until convergence [7]. Researchers have been studying how to distribute
ANN training to mitigate this. Distributed ANNs reduce the time it takes to
train a complex ANN model by distributing its computation across multiple
compute nodes. This distribution can follow different parallelization methods,
system architectures, and synchronization settings [38].

####  II-C1 Parallelization methods

There are many possible parallelization strategies for ANNs [38]. We describe
below the two most common ones: model and data parallelism.

Model Parallelism.  In model parallelism [18], the ANN model is split into
different parts which are distributed among the worker nodes. The major
challenge when using model parallelism is to determine how to partition the
model and keep the computation balanced among the workers [38]. Considering
the inherent difficulty of developing a splitting method that is generic
enough to be used on different ANN models and scalable enough to be
distributed across multiple nodes, we omit the consideration of model
parallelism in this work.

Data Parallelism.  In data parallelism [18], the $N$ training pairs within a
mini-batch are assumed to be independent of each other [39]. Thus, workers are
initialized as replicas and trained with different splits of those $N$ pairs.
The replicas perform the GC steps over the data received and then synchronize
their parameters among themselves. This method of parallelism is the most
common one and has existed since the first implementations of ANNs [38].
Frameworks like TensorFlow [12] or PyTorch [40] use data parallelism by
default when deployed on multi-core compute nodes or GPUs.

####  II-C2 System Architectures

The synchronization among the workers’ parameters in a data-parallel ANN
setting is either centralized or decentralized [38], as described below.

Centralized.  In a centralized architecture [18], workers systematically send
their parameter updates to one or multiple parameter servers. Those servers
aggregate the updates of all workers and apply them to a centralized model
[38]. Then, the workers use the centralized model on the next iterations of
BGD. Despite being easier to manage compared to decentralized architectures,
the scalability when using parameter servers is limited. Thus, by relying on
parameter servers to aggregate updates and broadcast them to all workers, the
parameter servers may become the actual bottleneck of such an architecture
[41].

Decentralized.  In a decentralized architecture [18], the workers synchronize
themselves using a broadcasting-like form of communication [42] which
eliminates the bottleneck of the parameter servers. An AllReduce operation may
be performed on a fully connected network, generating a communication cost of
$\mathbf{O}(n^{2})$ over $n$ workers; or, on ring-like topologies, an Ring-
AllReduce operation may reduce the communication to $\mathbf{O}(n)$ [38] but
typically increases the time it takes to propagate the parameters through the
network.

####  II-C3 Synchronization Settings

The aggregation and application of the model updates in a data-parallel ANN
system can be synchronous or asynchronous. Those settings are described below.

Synchronous.  In a synchronous setting [18], workers have to synchronize
themselves after each iteration. Therefore, they can only initialize the next
GC step if their models are synchronized. This synchronization barrier
directly facilitates the proof of convergence of SGD (as in a centralized
setting) but wastes computational resources at idle times (i.e., when workers
have to wait for others to resume their computation) [38].

Asynchronous.  In an asynchronous SGD (ASGD) setting [18], workers are allowed
to compute GC steps also on stale model parameters. This behaviour obviously
minimizes idle times but makes it harder to mathematically prove SGD
convergence. Recent developments on ASGD [19, 25, 20, 24, 27, 43, 21, 22, 23],
however, have tackled exactly this issue under different assumptions.
Koloskova et al. [19] recently proved that ASGD is always faster than BGD.
Furthermore, they also show that ASGD converges under a parameter-server
setting within
$\mathcal{O}(\frac{\sigma^{2}}{\epsilon^{2}}+\frac{\sqrt{\tau_{max}\tau_{avg}}}{\epsilon})$
iterations to an $\epsilon$-small error, with $\tau$ being the gradient delay
and a bounded variance for the gradients $\sigma^{2}\geq 0$.

###  II-D TensorFlow

TensorFlow (TF) provides a collection of APIs to implement and execute ML
algorithms [12]. It is one of the most popular ANN frameworks and is used by
companies such as Google, Intel, Twitter, and Coca-Cola. TensorFlow is open-
source, and it is available in multiple languages such as Python, Java, C,
C++, and Go. Noteworthy TF features are its performance, transparent
acceleration via GPUs, and its integration with Keras [44]. Keras is a high-
level framework, written in Python, that simplifies ANN implementations. TF
supports distributed training via the tf.distribute.Strategy API. This API
allows the user to choose among multiple distributed strategies, such as the
MirrorredStrategy (referred to as “TF_Mirror” in our experiments), which is
decentralized and synchronous, and the ParameterServerStrategy, which is
centralized and can be either synchronous or asynchronous.

###  II-E AIR Distributed Dataflow Engine

Asynchronous Iterative Routing (AIR) is a distributed dataflow engine which
implements a light-weight iterative sharding protocol [45] on top of the
Message Passing Interface (MPI). It is a native stream-processing engine that
processes complex dataflows, consisting of direct acyclic graphs (DAGs) of
logical dataflow operators. AIR’s main features are that it uses an
asynchronous MPI protocol and does not rely on a master-client architecture
(but follows a pure client-client pattern) for communication. Both of these
characteristics differ AIR both from existing bulk-synchronous processing
(BSP) systems, such as Apache Spark [30], and asynchronous stream-processing
(ASP) engines, such as Apache Flink [8] and Samza [31]. AIR is implemented in
C++ using POSIX threads (pthreads) for multi-threading and MPI for
communication among nodes. Due to its light-weight and robust architecture,
AIR’s scale-out performance has been shown to be up to 15 times better than
Spark and up to 9 times better than Flink [28], on a variety of HPC settings.

##  III TensAIR Architecture

We now introduce the architecture of TensAIR, a distributed stream-processing
engine (supporting the common dataflow operators like Map, Reduce, Join,
etc.), which has been augmented with the data-parallel, decentralized,
asynchronous ANN operator Model, with train and predict as two new OL
functions. TensAIR is a TensorFlow framework developed on top of AIR [45, 28],
which enables a remarkable scale-out performance for training ANNs in an
online learning scenario.

###  III-A TensAIR

Just like AIR, TensAIR is a native stream-processing engine that processes
complex dataflows consisting of logical dataflow operators. This means that
TensAIR can scale out both the training and prediction tasks of an ANN model
to multiple compute nodes, either with or without GPUs associated with them.
TensAIR has its dataflow defined based on AIR operators and can be visualized
using a graph (see Figure 1). However, different from AIR, a TensAIR dataflow
is not a DAG because it allows cycles among the model operators. Other than
some adaptations to the source code to allow cycles in an AIR dataflow, our
TensAIR extension did however not affect the general AIR framework (nor
TensorFlow itself).

Figure 1: TensAIR dataflow with $n$ distributed ModelW2V instances and a
single instance of Map, Split, UDF and ModelSA.

Figure 1 depicts the dataflow for a Sentiment Analysis (SA) use-case on a
Twitter data stream. This dataflow predicts the sentiments of live tweets
using a pre-trained ANN model (ModelSA). However, it does not rely on pre-
defined word embeddings. The dataflow constantly improves its embeddings on a
second Word2Vec (W2V) ANN model (ModelW2V), which it trains using the same
input stream as used for the predictions. By following a passive concept-drift
adaptation strategy, it can adapt its sentiment predictions in real-time based
on changing word distributions among the input tweets. Moreover, it does not
require any sentiment labels for newly streamed tweets at ModelSA, since only
ModelW2V is re-trained in a self-supervised manner by generating mini-batches
of word pairs $(x,y)$ directly from the input tweets.

TensAIR Dataflows.  Generally, a TensAIR dataflow is composed of logical
dataflow operators which all extend a basic Vertex superclass in AIR [45].
Vertex implements AIR’s asynchronous MPI protocol via multi-threaded queues of
incoming and outgoing messages, which are exchanged among all nodes (aka.
“ranks”) in the network asynchronously. The number of instances of each Vertex
subclass can be configured beforehand. In Figure 1, we represent ModelW2V with
n instances, while Map, Split, UDF and ModelSA have 1 instance here for
simplicity.

Our SA dataflow starts with Map which receives tweets from a Twitter input
stream (implemented via cURL or a file interface) and tokenizes the tweets
based on the same word dictionary also used by ModelW2V and ModelSA. Split
then identifies whether the tokenized tweets shall be used for re-training the
word embeddings, for sentiment prediction, or for both. If the tokenized
tweets are selected for training, they are turned into mini-batches via the
UDF operator. The $(x,y)$ word pairs in each mini-batch $X$ are sharded across
Model${}^{W2V}_{1}$, $\ldots$, Model${}^{W2V}_{n}$ with a standard hash-
partitioner using words $x$ as keys. ModelW2V implements a default skip-gram
model. If the tokenized tweets are selected for prediction, a tweet is
vectorized by using the word embeddings obtained from any of the ModelW2V
instances (as discussed in Section III-C, they will all eventually converge to
the same common model) and sent to the pre-trained ModelSA which then predicts
the tweets’ sentiments as either positive or negative. All ModelW2V instances
are initialized with a copy of the same TensorFlow model, which can be pre-
configured in Keras and be loaded from a file into TensAIR.

Stream Processing.  As shown in Algorithm 1, a TensAIR Model operator has two
new OL functions train and predict, which can asynchronously send and receive
messages to and from other operators. For ModelSA, predict receives embedded
tweets as input, which are obtained from the current parameters of any of the
ModelW2V instances. For a ModelW2V, train receives either encoded mini-batches
$X$ or gradients $\nabla x$ as messages. Each message encoding a gradient that
was computed by another model instance is immediately used to update the local
model accordingly. Each mini-batch first invokes a local gradient computation
and is then used to update the local model. Each such resulting gradient is
also locally buffered until a desired buffer size ($\mathit{maxGradBuffer}$)
for the outgoing gradients is reached, upon which the buffer then is broadcast
to all other model instances.

1 Constructor  _Model(_$\mathit{tfModel}$ , $\mathit{maxBuffer}$_) extends
Vertex_:

2 model = $\mathit{tfModel}$;

3 maxGradBuffer = $\mathit{maxBuffer}$;

4 gradients = $\emptyset$;

5 ALIVE = $\mathit{true}$;

6 Function  _streamProcess(_$\mathit{msg}$_)_:

7 while  _ALIVE_ do

8 if  _$\mathit{msg}.\mathit{mode}$ == TRAIN_ then

9 train($\mathit{msg}$);

else

10 predict($\mathit{msg}$);

end if

end while

11 Function  _train(_$\mathit{msg}$_)_:

12 if  _$\mathit{msg}.\mathit{isGradient}$_ then

13 model = apply_gradient(model, $\mathit{msg}$);

else

14 gradient = calculate_gradient(model, $\mathit{msg}$);

15 model = apply_gradient(model, gradient);

16 gradients = gradients $\cup$ $\\{$gradient$\\}$;

17 if  _$|$ gradients$|$ $\geq$ maxGradBuffer_ then

18 send_gradients(gradients);

end if

end if

19 Function  _predict(_$\mathit{msg}$_)_:

20 predictions = model.make_predictions($\mathit{msg}$)

21 send_results(predictions)

Algorithm 1 TensAIR Model class with additional OL functions train and predict
(pseudocode)

###  III-B Model Consistency

Despite TensAIR’s asynchronous nature, it is necessary to maintain the models
consistent among themselves during training in order to guarantee that they
are aligned and, therefore, they eventually convergence to a same common
model. In TensAIR, this is given by the exchange of gradients between the
various Model instances.

Due to our asynchronous computation and application of the gradients on the
distributed model instances, Model${}^{W2V}_{i}$ receives gradients calculated
by Model${}^{W2V}_{j}$ (with $j\neq i$) which are similar but not necessarily
equal to itself. This occurs whenever Model${}^{W2V}_{i}$, which has already
applied to itself a set of $G_{i}=\\{\nabla x,\nabla y,...,\nabla z\\}$
gradients, calculates a new gradient $\nabla a$, and sends it to
Model${}^{W2V}_{j}$, such that $G_{i}\neq G_{j}$ at the time when
Model${}^{W2V}_{j}$ applies $\nabla a$. The difference $|G_{i}\cup
G_{j}|-|G_{i}\cap G_{j}|$ between these two models is defined as staleness
[46]. This $\mathit{staleness}_{i,j}(\nabla_{a})$ metric is the symmetric
distance between $G_{i}$ and $G_{j}$ with respect to the times at which a new
gradient $\nabla_{a}$ was computed by a model $i$ and is applied to model $j$,
respectively. This phenomenon and the staleness metric are illustrated in
Figure 2.

Figure 2: ASGD staleness on a distributed framework.

Figure 2 illustrates the timeline of messages (containing both mini-batches
and gradients) exchanged among TensAIR models considering
$\mathit{max}\\-\mathit{GradBuffer}=1$. Assume the UDF distributes 5 mini-
batches to 3 models. After receiving their first mini-batch, each
Model${}^{W2V}_{i}$ calculates a corresponding gradient. Note that, when
applied locally, the staleness of any gradient is 0 because it is computed and
immediately applied by the same model. While computing or applying a local
gradient, each Model${}^{W2V}_{i}$ may receive more gradients to calculate
and/or apply from either the UDF or other models asynchronously. In our
protocol, the models first finish their current gradient computation,
immediately apply it locally, then buffer and send $\mathit{maxGradBuffer}$
many locally computed gradients to the other models, and wait for their next
update.

As illustration, take a look at Model1 in Figure 2. While computing
$\nabla_{blue}$, it receives the yellow mini-batch from the UDF, which it
starts computing immediately after it finishes processing the blue one—which
it had already started when it received the yellow mini-batch. During the
computation of $\nabla_{yellow}$, Model1 receives $\nabla_{green}$ to apply,
which it does promptly after finishing $\nabla_{yellow}$. Note that when
Model2 computed $\nabla_{green}$ and Model3 computed $\nabla_{red}$, they have
not applied a single gradient to their local models at that time. Thus,
$|G_{2}|=|G_{3}|=0$. However, before applying $\nabla_{green}$,
$G_{1}=\\{\nabla_{blue},\nabla_{yellow}\\}$ with $|G_{1}|=2$ and
$\mathit{staleness}_{1,2}(\nabla_{green})=2$. Along the same lines, before
applying $\nabla_{red}$, $|G_{1}|=3$ and
$\mathit{staleness}_{1,3}(\nabla_{red})=3$.

###  III-C Model Convergence

Since TensAIR operates on data streams and is both asynchronous and fully
decentralized (i.e., it has no centralized parameter server), it exhibits
characteristics which current SGD proofs of convergence [23, 21, 20] do not
cover. Therefore, we next discuss under which circumstances TensAIR is
guaranteed to converge.

First, we consider that training is performed between significant concept
drifts. Therefore, we assume that the mini-batches between two subsequent
concept drifts do not change. Thus, if a concept drift occurs during the
training, the model will not converge until the concept drift ends. By
considering this, the data stream between two concept drifts will behave like
a fixed data set. In this case, if given enough training examples, as seen in
[35], each of the local model instances will eventually converge.

Second, considering TensAIR’s asynchronous and distributed nature, our SGD
updates can be staled. We are not aware of any ASGD proofs of convergence that
contemplate staleness in a decentralized setting. However, Koloskova et al.
[19] recently proved a tighter convergence rate for ASGD in a parameter-server
setting with varying computation and communication frequency among the worker
nodes. Specifically, they show ASGD convergence within
$\mathcal{O}(\frac{\sigma^{2}}{\epsilon^{2}}+\frac{\sqrt{\tau_{max}\tau_{avg}}}{\epsilon})$
iterations to an $\epsilon$-small error, with $\tau$ being the gradient delay
and a bounded variance for the gradients $\sigma^{2}\geq 0$. Although TensAIR
does not rely on a centralized parameter server, its convergence rate can be
the same as the one from a parameter-server setting under specific conditions.
One may achieve this by introducing global synchronization steps to ensure
that all local models become synchronized periodically, a strategy called
Stale-Synchronous Parallelism (SSP) [39]. Under the SSP strategy, the local
gradients can still be computed asynchronously but the models updates are
limited to a synchronous communication. We note that staleness under SSP is
analogous to the gradient delay $\tau$ in [19]. Both $\tau$ and staleness
measure the difference between the state of the model in which a gradient was
calculated versus in which it is applied. Considering this, we can reduce the
SSP convergence proof to a special case of the parameter-server based proof
from [19], in which all worker nodes from the parameter server are updated
simultaneously.

In practice, we however did not observe any degradation in the convergence
rate of TensAIR when compared to synchronous SGD (see Section IV). Therefore,
we did not further explore the SSP strategy due to its performance limitations
when compared to a fully asynchronous system.

###  III-D Implementation

TensAIR is completely implemented in C++. It includes the TensorFlow 2.8
native C API to load, save, train, and predict ANN models. Therefore, it is
possible to develop a TensorFlow/Keras model in Python, save the model to a
file, and load it directly into TensAIR. TensAIR is completely open-source and
available from our Gitlab repository111https://gitlab.uni.lu/mdalle/TensAIR.

##  IV Experiments & Discussion

To assess TensAIR, we performed experiments to measure its performance on
solving prototypical ML problems such as Word2Vec (word embeddings) and
CIFAR-10 (image classification). We empirically validate TensAIR’s model
convergence by comparing its training loss curve at increasing levels of
distribution across both CPUs and GPUs. Our results confirm that TensAIR’s
ASGD updates achieve similar convergence on Word2Vec and CIFAR-10 as a
synchronous SGD propagation. At the same time, we achieve a nearly linear
reduction in training time on both problems. Due to this reduction, TensAIR
significantly outperforms not just the Apache Kafka and Flink extensions
(based on both the standard and distributed TensorFlow APIs), but also Horovod
which is a long-standing effort to scale-out ANN training. Finally, by
providing an in-depth analysis of a sentiment analysis (SA) use-case on
Twitter, we demonstrate the importance of OL in the presence of concept drifts
(i.e., COVID-19 related tweets with changing sentiments). In particular the SA
usecase is an example of task that would be deemed too complex to be adapted
in real-time (at a throughput rate of ca. 6,000 tweets per second) when using
other OL frameworks.

HPC Setup.  We carried out the experiments described in this section using the
HPC facilities of the University of Luxembourg [47]. We distributed the ANNs
training using up to 4 Nvidia Tesla V100 GPUs per node with 768 GB RAM each.
We also deployed up to 16 regular nodes, with 28 CPU cores and 128 GB RAM
each, for the CPU-based (i.e., without using GPU acceleration) settings.

Event Generation.  We trained both sparse (word
embeddings222https://www.tensorflow.org/tutorials/text/word2vec) and dense
(image classification333https://www.tensorflow.org/tutorials/images/cnn)
models based on English Wikipedia articles and images from CIFAR-10 [48],
respectively. Instead of connecting to actual streams, we chose those static
datasets to facilitate a consistent analysis of the results and ensure
reproducibility. Moreover, to simulate a streaming scenario, we implemented an
additional MiniBatchGenerator as an entry-point Vertex operator (compare to
Figure 1) which generates events $e_{i}$ with timestamps $s_{i}$, groups them
into mini-batches $X_{j}$ by using a tumbling-window semantics, and sends
these mini-batches to the subsequent operators in the dataflow. Furthermore,
this allows us to simulate streams of unbounded size by iterating over the
datasets multiple times (in analogy to training with multiple epochs over a
fixed dataset).

Sparse vs. Dense Models.  We chose Word2Vec and CIFAR-10 because they
represent prototypical ML problems with sparse and dense model updates,
respectively. Sparse updates mean that only a small portion of the neural
network variables actually become updated per mini-batch [49]. Hence,
sparseness should assist the model convergence when using ASGD, as observed
also in Hogwild! [49]. We trained by sampling 1% from English Wikipedia which
corresponds to 11.7M training examples (i.e., word pairs). On the other hand,
we chose CIFAR-10 for being dense. Thus, we could analyze how this
characteristic possibly hinders convergence when models are distributed and
updated asynchronously. We train on all of the 50,000 labeled images of the
CIFAR-10 dataset.

###  IV-A Convergence Analysis

We first explored TensAIR’s ability to converge by determining if and how ASGD
might degrade the quality of the trained model (Figure 3). We compared the
training loss curve of Word2Vec and CIFAR-10 by distributing TensAIR models
from 1 to 4 GPUs using 1 TensAIR rank per GPU (Figures 3(b) & 3(d)). We
additionally explored the models convergence when trained with distributed CPU
nodes (Figures 3(a) & 3(c)). In this second scenario, we trained up to 64
ranks on 16 nodes simultaneously without GPUs. Note that, when using a single
TensAIR rank, TensAIR’s gradient updates behave as in a synchronous SGD
implementation.

((a))

((b))

((c))

((d))

Figure 3: Convergence analysis of TensAIR on the Word2Vec and CIFAR-10 use-
cases.

((a))

((b))

((c))

((d))

Figure 4: Speedup analysis of TensAIR on the Word2Vec and CIFAR-10 use-cases.

The extremely low variance among all loss curves shown in Figures 3(a) and
3(b) demonstrates that our asynchronous and distributed SGD updates do not at
all negatively affect the convergence of the Word2Vec models. We assume that
this is due to (1) the sparseness of Word2Vec, and (2) a low staleness of the
gradients (which are relatively inexpensive to compute and apply for
Word2Vec). The low staleness indicates a fast exchange of gradients among
models.

In Figure 3(c), we however observe a remarkable degradation of the loss when
distributing CIFAR-10 across multiple nodes. This is due to the fixed learning
rate used on all settings being the same. When distributing dense models on
multiple ranks without adapting the mini-batch size, it is well known to
result in a degradation of the loss curve (even on synchronous settings). This
degradation occurs because the behaviour of training $N$ models with mini-
batches of size $x$ is similar to training 1 model with mini-batches of size
$N\cdot x$. To mitigate this issue, Horovod recommends to increase the
learning rate $\alpha$ by the number of ranks used to distribute the model
[50], i.e., $\alpha_{new}=\alpha\cdot N$. Accordingly, in Figure 3(d), we
again do not see any degradation of the loss when distributing CIFAR-10 across
multiple GPUs because we use a maximum of 4 GPUs.

###  IV-B Speed-up Analysis

Next, we explore the performance of TensAIR under increasing levels of
distribution and with respect to varying mini-batch sizes over both Word2Vec
and CIFAR-10. This experiment is also deployed on up to 64 ranks (16 nodes)
and up to 4 GPUs (1 node). We observe in Figure 4 that TensAIR achieves a
nearly-linear scale-out under most of our settings. In most cases, TensAIR
achieves a better speedup when training with smaller mini-batches. This
difference is because the mini-batch size is inversely proportional to the
training time. Hence, the smaller the training time, the bigger is the
fraction of the computation that is not distributed. Thus, models with
expensive gradient computations will have a better scale-out performance.

###  IV-C Baseline Comparison

Apart from TensAIR, it is also possible to train ANNs by using Apache Kafka
and Flink as message brokers to generate data streams of varying throughputs.
Kafka is already included in the standard TensorFlow I/O library
(tensorflow_io), which however allows no actual distribution in the training
phase [10]. Flink, on the other hand, employs the distributed TensorFlow API
(tensorflow.distribute). However, we were not able to run the provided dl-on-
flink use-case [6] even after various attempts on our HPC setup. We therefore
report the direct deployment of our Word2Vec and CIFAR-10 use-cases (Figures 5
& 6) on both the standard and distributed TensorFlow APIs (the latter using
the MirroredStrategy option of tensorflow.distribute). We thereby, simulate a
streaming scenario by feeding one mini-batch per training iteration into
TensorFlow, which yields a very optimistic upper-bound for the maximum
throughput that Kafka and Flink could achieve. In a similar manner, we also
determined the maximum throughput of Horovod [7], which is however not a
streaming engine by default.

Figure 5: Throughput comparison between TensAIR, TensorFlow (standard and
distributed), and Horovod in the W2V use-case. Figure 6: Throughput comparison
between TensAIR, TensorFlow (standard and distributed), and Horovod in the
CIFAR-10 use-case.

In Figures 5 and 6, we see that TensAIR clearly surpasses both the standard
and distributed TensorFlow setups as well as Horovod. This occurs because, as
opposed to TensAIR, their architectures were not developed to train on batches
arriving from data streams. Thus, in a streaming scenario, the overhead of
transferring the training data to the worker nodes increases by the number of
training steps. On the other hand, TensAIR is an end-to-end dataflow engine
prepared to train ANNs from streaming data. Thus, the transfer of training
data overhead is mitigated by the asynchronous iterative routing protocol.
This allows TensAIR to (1) reduce both computational resources and idle times
while the data is being transferred, and (2) have an optimized buffer
management for incoming mini-batches and outgoing gradients, respectively.

In our experiments, we could sustain a maximum training rate of 285,560
training examples per second on Word2Vec and 200,000 images per second on
CIFAR-10, which corresponds to sustainable throughputs of 14.16 MB/s and 585
MB/s respectively. We reached these values by training with 3 GPUs on Word2Vec
and 4 GPUs on CIFAR-10.

###  IV-D Sentiment Analysis of COVID19

Here, we exemplify the benefits of training an ANN in real-time from streaming
data. To this end, we analyze the impact of concept drifts on a sentiment
analysis setting, specifically drifts that occurred during and due to the
COVID19 pandemic. First, we trained a large Word2Vec model using 20% of
English Wikipedia plus the Sentiment140 dataset [51] from Stanford. Then, we
trained an LSTM model [52] using the Sentiment140 dataset together with the
word embeddings we trained previously. After three epochs, we reached 78%
accuracy on the training and the test set. However, language is always
evolving. Thus, this model may not sustain its accuracy for long if deployed
to analyze streaming data in real-time. We exemplify this by fine-tuning the
word embeddings with 2M additional tweets published from November 1st, 2019 to
October 10th, 2021 containing the following keywords: covid19, corona,
coronavirus, pandemic, quarantine, lockdown, sarscov2. Then, we compared the
previously trained word embeddings and the fine-tuned ones and found an
average cosine difference of only 2%. However, despite being small, this
difference is concentrated onto specific keywords.

Term | rt | corona | pandemic | booster | 2021
---|---|---|---|---|---
Difference | 0.728 | 0.658 | 0.646 | 0.625 | 0.620
TABLE I: Cosine differences after updating word embeddings.

As shown in Table I, keywords related to the COVID-19 pandemic are the ones
that most suffered from a concept drift. Take as example pandemic, booster and
corona, which had over 62% of cosine difference before and after the Word2Vec
models have been updated. Due to the concept drift, the sentiment over
specific terms and, consequently, entire tweets also changed. One observes
this change by comparing the output of our LSTM model when: (1) inputting
tweets embedded with the pre-trained word embeddings; (2) inputting tweets
embedded with the fine-tuned word embeddings. Take as an example the sentence
“I got corona.”, which had a sentiment of $+2.0463$ when predicted with the
pre-trained embeddings; and $-2.4873$ when predicted with the fine-tuned
embeddings. Considering that the higher the sentiment value the more positive
the tweet is, we can observe that corona (also representing a brand of a beer)
was seen as positive and now is related to a very negative sentiment.

To tackle concept drifts in this use-case, we argue that TensAIR with its OL
components (as depicted in Figure 1) could be readily deployed. A real-time
pipeline with Twitter would allow us to constantly update the word embeddings
(our sustainable throughput would be more than sufficient compared to the
estimated throughput of Twitter). Consequently, the sentiment analysis
algorithm would always be up-to-date with respect to such concept drifts.

##  V Conclusions

OL is an emerging area of research which still has not extensively explored
the real-time training of ANNs. In this paper, we introduced TensAIR, a novel
end-to-end dataflow engine for OL from data streams. It uses an asynchronous
iterative routing (AIR) protocol to train and predict ANNs in a distributed
manner. One major feature of TensAIR is its high performance due to its direct
inclusion of TensorFlow as a first-class dataflow operator. TensAIR achieves a
nearly linear scale-out performance in terms of sustainable throughput and
with respect to its number of worker nodes. It uses TensorFlow models and
supports the common dataflow operators, such as Map, Reduce, and Join, which
facilitates the implementation of diverse use-cases. Therefore, we highlight
the following capabilities of TensAIR: (1) process multiple data streams
simultaneously; (2) train models using either CPUs, GPUs, or both; (3) train
ANNs in an asynchronous and distributed manner; and (4) incorporate user-
defined data pre- and post-processing pipelines. We empirically demonstrate
that, on a real-time streaming scenario, TensAIR surpasses both the standard
and distributed TensorFlow APIs (representing upper bounds for the throughput
of Apache Kafka and Flink, respectively) when training ANNs from streaming
data. We also show that, in our experiments, the model degradation due to ASGD
is insignificant in practice.

As future work, we believe that TensAIR may also lead to novel online learning
use-cases which were previously considered too complex but now become feasible
due to very good scale-out performance of TensAIR. Specifically, we intend to
study similar learning tasks over audio/video streams which we see as the main
target domain for stream processing. To reduce the computational cost of
training an ANN indefinitely, we shall also investigate how different active
concept-drift detection algorithms behave under an OL setting with ANNs.

## Acknowledgment

This work is funded by the Luxembourg National Research Fund under the PRIDE
program (PRIDE17/12252781).

## References

  * [1] S. C. Hoi, D. Sahoo, J. Lu, and P. Zhao, “Online learning: A comprehensive survey,” _Neurocomputing_ , vol. 459, pp. 249–289, 2021.
  * [2] A. S. Iwashita and J. P. Papa, “An overview on concept drift learning,” _IEEE access_ , vol. 7, pp. 1532–1547, 2018.
  * [3] M. Heusinger, C. Raab, and F.-M. Schleif, “Passive concept drift handling via variations of learning vector quantization,” _Neural Computing and Applications_ , pp. 1–12, 2020.
  * [4] S. Afaq and S. Rao, “Significance of epochs on training a neural network,” _International Journal of Scientific and Technology Research_ , vol. 19, no. 6, pp. 485–488, 2020.
  * [5] J. Gama, I. Žliobaitė, A. Bifet, M. Pechenizkiy, and A. Bouchachia, “A survey on concept drift adaptation,” _ACM computing surveys (CSUR)_ , vol. 46, no. 4, pp. 1–37, 2014.
  * [6] (2022) Deep learning on Flink. Github. Accessed: 2022-08-05. [Online]. Available: https://github.com/flink-extended/dl-on-flink
  * [7] A. Sergeev and M. Del Balso, “Horovod: fast and easy distributed deep learning in Tensorflow,” _arXiv preprint arXiv:1802.05799_ , 2018.
  * [8] P. Carbone, A. Katsifodimos, S. Ewen, V. Markl, S. Haridi, and K. Tzoumas, “Apache Flink: Stream and batch processing in a single engine,” _Bulletin of the IEEE Computer Society Technical Committee on Data Engineering_ , vol. 36, no. 4, 2015.
  * [9] J. Kreps, N. Narkhede, J. Rao _et al._ , “Kafka: A distributed messaging system for log processing,” in _Proceedings of the NetDB_ , vol. 11, 2011, pp. 1–7.
  * [10] (2022) Robust machine learning on streaming data using Kafka and Tensorflow-IO. Tensorflow I/O tutorials. Accessed: 2022-08-05. [Online]. Available: https://www.tensorflow.org/io/tutorials/kafka
  * [11] M. Boehm, M. W. Dusenberry, D. Eriksson, A. V. Evfimievski, F. M. Manshadi, N. Pansare, B. Reinwald, F. R. Reiss, P. Sen, A. C. Surve _et al._ , “SystemML: Declarative machine learning on spark,” _Proceedings of the VLDB Endowment_ , vol. 9, no. 13, pp. 1425–1436, 2016.
  * [12] M. Abadi, P. Barham, J. Chen, Z. Chen, A. Davis, J. Dean, M. Devin, S. Ghemawat, G. Irving, M. Isard _et al._ , “Tensorflow: a system for large-scale machine learning,” in _12th USENIX symposium on operating systems design and implementation (OSDI 16)_ , 2016, pp. 265–283.
  * [13] J. Lu, A. Liu, F. Dong, F. Gu, J. Gama, and G. Zhang, “Learning under concept drift: A review,” _IEEE Transactions on Knowledge and Data Engineering_ , vol. 31, no. 12, pp. 2346–2363, 2018.
  * [14] H. Hu, M. Kantardzic, and T. S. Sethi, “No free lunch theorem for concept drift detection in streaming data classification: A review,” _Wiley Interdisciplinary Reviews: Data Mining and Knowledge Discovery_ , vol. 10, no. 2, p. e1327, 2020.
  * [15] S. Priya and R. A. Uthra, “Deep learning framework for handling concept drift and class imbalanced complex decision-making on streaming data,” _Complex & Intelligent Systems_, pp. 1–17, 2021.
  * [16] R. S. M. Barros and S. G. T. C. Santos, “A large-scale comparison of concept drift detectors,” _Information Sciences_ , vol. 451, pp. 348–370, 2018.
  * [17] T. S. Sethi and M. Kantardzic, “On the reliable detection of concept drift from streaming unlabeled data,” _Expert Systems with Applications_ , vol. 82, pp. 77–99, 2017.
  * [18] S. Ouyang, D. Dong, Y. Xu, and L. Xiao, “Communication optimization strategies for distributed deep neural network training: A survey,” _Journal of Parallel and Distributed Computing_ , vol. 149, pp. 52–65, 2021.
  * [19] A. Koloskova, S. U. Stich, and M. Jaggi, “Sharper convergence guarantees for asynchronous SGD for distributed and federated learning,” in _Advances in Neural Information Processing Systems_ , A. H. Oh, A. Agarwal, D. Belgrave, and K. Cho, Eds., 2022.
  * [20] J. Jiang, B. Cui, C. Zhang, and L. Yu, “Heterogeneity-aware distributed parameter servers,” in _Proceedings of the 2017 ACM International Conference on Management of Data_ , 2017, pp. 463–478.
  * [21] T. Sun, R. Hannah, and W. Yin, “Asynchronous coordinate descent under more realistic assumptions,” _Advances in Neural Information Processing Systems_ , vol. 30, 2017.
  * [22] X. Lian, Y. Huang, Y. Li, and J. Liu, “Asynchronous parallel stochastic gradient for nonconvex optimization,” _Advances in Neural Information Processing Systems_ , vol. 28, 2015.
  * [23] X. Zhang, J. Liu, and Z. Zhu, “Taming convergence for asynchronous stochastic gradient descent with unbounded delay in non-convex learning,” in _2020 59th IEEE Conference on Decision and Control (CDC)_ , 2020, pp. 3580–3585.
  * [24] W. Dai, A. Kumar, J. Wei, Q. Ho, G. Gibson, and E. Xing, “High-performance distributed ml at scale through parameter server consistency models,” in _Proceedings of the AAAI Conference on Artificial Intelligence_ , vol. 29, no. 1, 2015.
  * [25] J. Dean, G. Corrado, R. Monga, K. Chen, M. Devin, M. Mao, M. Ranzato, A. Senior, P. Tucker, K. Yang _et al._ , “Large scale distributed deep networks,” _Advances in neural information processing systems_ , vol. 25, 2012.
  * [26] H. Zhang, C.-J. Hsieh, and V. Akella, “Hogwild++: A new mechanism for decentralized asynchronous stochastic gradient descent,” in _2016 IEEE 16th International Conference on Data Mining (ICDM)_. IEEE, 2016, pp. 629–638.
  * [27] J. Liu, S. Wright, C. Ré, V. Bittorf, and S. Sridhar, “An asynchronous parallel stochastic coordinate descent algorithm,” in _International Conference on Machine Learning_. PMLR, 2014, pp. 469–477.
  * [28] V. E. Venugopal, M. Theobald, D. Tassetti, S. Chaychi, and A. Tawakuli, “Targeting a light-weight and multi-channel approach for distributed stream processing,” _Journal of Parallel and Distributed Computing_ , 2022.
  * [29] H. M. Gomes, J. Read, A. Bifet, J. P. Barddal, and J. Gama, “Machine learning for streaming data: state of the art, challenges, and opportunities,” _ACM SIGKDD Explorations Newsletter_ , vol. 21, no. 2, pp. 6–22, 2019.
  * [30] M. Zaharia, R. S. Xin, P. Wendell, T. Das, M. Armbrust, A. Dave, X. Meng, J. Rosen, S. Venkataraman, M. J. Franklin _et al._ , “Apache Spark: a unified engine for big data processing,” _Communications of the ACM_ , vol. 59, no. 11, pp. 56–65, 2016.
  * [31] S. A. Noghabi, K. Paramasivam, Y. Pan, N. Ramesh, J. Bringhurst, I. Gupta, and R. H. Campbell, “Samza: stateful scalable stream processing at LinkedIn,” _Proceedings of the VLDB Endowment_ , vol. 10, no. 12, pp. 1634–1645, 2017\.
  * [32] T. Akidau, R. Bradshaw, C. Chambers, S. Chernyak, R. J. Fernández-Moctezuma, R. Lax, S. McVeety, D. Mills, F. Perry, E. Schmidt, and S. Whittle, “The dataflow model: A practical approach to balancing correctness, latency, and cost in massive-scale, unbounded, out-of-order data processing,” _Proc. VLDB Endow._ , vol. 8, no. 12, p. 1792–1803, aug 2015\. [Online]. Available: https://doi.org/10.14778/2824032.2824076
  * [33] J. Karimov, T. Rabl, A. Katsifodimos, R. Samarev, H. Heiskanen, and V. Markl, “Benchmarking distributed stream data processing systems,” in _2018 IEEE 34th International Conference on Data Engineering (ICDE)_ , 2018, pp. 1507–1518.
  * [34] S. Kulkarni, N. Bhagat, M. Fu, V. Kedigehalli, C. Kellogg, S. Mittal, J. M. Patel, K. Ramasamy, and S. Taneja, “Twitter heron: Stream processing at scale,” in _Proceedings of the 2015 ACM SIGMOD International Conference on Management of Data_ , ser. SIGMOD ’15. New York, NY, USA: ACM, 2015, p. 239–250.
  * [35] I. Goodfellow, Y. Bengio, and A. Courville, _Deep Learning_. MIT Press, 2016, http://www.deeplearningbook.org.
  * [36] S. Ruder, “An overview of gradient descent optimization algorithms,” _arXiv preprint arXiv:1609.04747_ , 2016.
  * [37] H. Robbins and S. Monro, “A stochastic approximation method,” _The annals of mathematical statistics_ , pp. 400–407, 1951.
  * [38] R. Mayer and H.-A. Jacobsen, “Scalable deep learning on distributed infrastructures: Challenges, techniques, and tools,” _ACM Computing Surveys (CSUR)_ , vol. 53, no. 1, pp. 1–37, 2020.
  * [39] T. Ben-Nun and T. Hoefler, “Demystifying parallel and distributed deep learning: An in-depth concurrency analysis,” _ACM Computing Surveys (CSUR)_ , vol. 52, no. 4, pp. 1–43, 2019.
  * [40] A. Paszke, S. Gross, F. Massa, A. Lerer, J. Bradbury, G. Chanan, T. Killeen, Z. Lin, N. Gimelshein, L. Antiga, A. Desmaison, A. Kopf, E. Yang, Z. DeVito, M. Raison, A. Tejani, S. Chilamkurthy, B. Steiner, L. Fang, J. Bai, and S. Chintala, “Pytorch: An imperative style, high-performance deep learning library,” in _Advances in Neural Information Processing Systems 32_ , H. Wallach, H. Larochelle, A. Beygelzimer, F. d'Alché-Buc, E. Fox, and R. Garnett, Eds. Curran Associates, Inc., 2019, pp. 8024–8035. [Online]. Available: http://papers.neurips.cc/paper/9015-pytorch-an-imperative-style-high-performance-deep-learning-library.pdf
  * [41] C. Chen, W. Wang, and B. Li, “Round-robin synchronization: Mitigating communication bottlenecks in parameter servers,” in _IEEE INFOCOM 2019-IEEE Conference on Computer Communications_. IEEE, 2019, pp. 532–540.
  * [42] S. Ouyang, D. Dong, Y. Xu, and L. Xiao, “Communication optimization strategies for distributed deep neural network training: A survey,” _Journal of Parallel and Distributed Computing_ , vol. 149, pp. 52–65, 2021.
  * [43] J. Liu and S. J. Wright, “Asynchronous stochastic coordinate descent: Parallelism and convergence properties,” _SIAM Journal on Optimization_ , vol. 25, no. 1, pp. 351–376, 2015.
  * [44] F. Chollet _et al._ (2015) Keras. [Online]. Available: https://github.com/fchollet/keras
  * [45] V. E. Venugopal, M. Theobald, S. Chaychi, and A. Tawakuli, “AIR: A light-weight yet high-performance dataflow engine based on asynchronous iterative routing,” in _2020 IEEE 32nd International Symposium on Computer Architecture and High Performance Computing (SBAC-PAD)_ , 2020, pp. 51–58.
  * [46] M. D. Tosi, V. Ellampallil Venugopal, and M. Theobald, “Convergence-time analysis of asynchronous distributed artificial neural networks,” in _5th Joint International Conference on Data Science & Management of Data (CODS/COMAD)_, 2022, pp. 314–315.
  * [47] S. Varrette, P. Bouvry, H. Cartiaux, and F. Georgatos, “Management of an academic hpc cluster: The ul experience,” in _Proc. of the 2014 Intl. Conf. on High Performance Computing & Simulation (HPCS 2014)_, 2014, pp. 959–967.
  * [48] A. Krizhevsky, G. Hinton _et al._ , “Learning multiple layers of features from tiny images,” University of Toronto, Department of Computer Science, Tech. Rep., 2009.
  * [49] B. Recht, C. Re, S. Wright, and F. Niu, “Hogwild!: A lock-free approach to parallelizing stochastic gradient descent,” _Advances in neural information processing systems_ , vol. 24, 2011.
  * [50] (2019) Horovod with keras. Horovod documentation. Accessed: 2022-05-18. [Online]. Available: https://horovod.readthedocs.io/en/stable/keras.html
  * [51] A. Go, R. Bhayani, and L. Huang, “Twitter sentiment classification using distant supervision,” _CS224N project report, Stanford_ , vol. 1, no. 12, 2009.
  * [52] (2022) Tensorflow. Tensorflow tutorials. Accessed: 2022-05-27. [Online]. Available: https://www.tensorflow.org/text/tutorials/text_classification_rnn