TensAIR: Real-Time Training of Neural Networks from Data-streams (2024)

2.1. Online Learning

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

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 (Carbone etal., 2015), Spark (Zaharia etal., 2016) and Samza (Noghabi etal., 2017), batches are distinguished into sliding windows, tumbling windows and (per-user) sessions (Akidau etal., 2015).

Latency vs. Throughput. When analyzing systems that process data streams, one typically benchmarks them by their latency and throughput (Karimov etal., 2018). 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” (Kulkarni etal., 2015), or even a crash).

Passive & Active Drift Adaptation. To adapt to concept drifts, one may rely on either passive or active adaptation strategies (Heusinger etal., 2020). 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.

2.2. Artificial Neural Networks

ANNs denote a family of supervised ML algorithms which are designed to be trained on a pre-defined dataset (Goodfellow etal., 2016). 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})}$.

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}$.

Distributed Artificial Neural Networks.Over the last years, ANN models have substantially 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 (Sergeev and DelBalso, 2018). 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 synchronisation settings (Mayer and Jacobsen, 2020).

The most common form of distributing ANNs, which we also use in this work, is referred to as data parallelism (Ouyang etal., 2021), in which workers are initialized with replicas of the same initial model and trained with disjoint splits of the training data. Moreover, the synchronisation among the workers’ parameters in a data-parallel ANN setting is either centralised or decentralised (Mayer and Jacobsen, 2020). In a centralised architecture (Ouyang etal., 2021), 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 centralised model (Mayer and Jacobsen, 2020). Thus, by relying on parameter servers to aggregate updates, the parameter servers may become the bottleneck of such an architecture (Chen etal., 2019). On the other hand, in a decentralised architecture (Ouyang etal., 2021), the workers synchronize themselves using a broadcasting-like form of communication (Ouyang etal., 2021). This broadcast eliminates the bottleneck of the parameter servers but requires a direct communication among worker nodes.

The parameter updates in a data-parallel ANN system can be synchronous or asynchronous. In a synchronous setting (Ouyang etal., 2021), workers have to synchronize themselves after each mini-batch iteration. This synchronization barrier wastes computational resources at idle times (i.e., when workers have to wait for others to resume their computation) (Mayer and Jacobsen, 2020). In an asynchronous SGD (ASGD) setting (Ouyang etal., 2021), workers are allowed to compute their gradient computations also on stale model parameters. This behaviour obviously minimizes idle times but makes it harder to mathematically prove SGD convergence. Recent developments on ASGD (Jiang etal., 2017; Sun etal., 2017; Zhang etal., 2020), however, have tackled exactly this issue under different assumptions. Zhang et al. (Zhang etal., 2020) recently proved an $\mathbf{o}(1/\sqrt{k})$ convergence rate for unbounded non-convex problems using ASGD under a centralised parameter server setup (where $k$ denotes the iteration among the ASGD updates). Additionally, (Lian etal., 2018; Bornstein etal., 2023; Tosi and Theobald, 2023) proved the convergence of ASGD on decentralized networks under distinct assumptions and network topologies.

3.1. 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${}_{i}$ receives gradients calculated by Model${}_{j}$ (with $j\neq i$) which are similar but not necessarily equal to itself. This occurs whenever Model${}_{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${}_{j}$, such that $G_{i}\neq G_{j}$ at the time when Model${}_{j}$ applies $\nabla a$. The difference $|G_{i}\cup G_{j}|-|G_{i}\cap G_{j}|$ between these two models is defined as staleness (Tosi etal., 2022). 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.We illustrate this phenomenon and the staleness metric in Figure 2.

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${}_{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${}_{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, apply it locally, then buffer and send $\mathit{maxGradBuffer}$ many locally computed gradients to the other models, and wait for their next update.

As an illustration, take a look at Model${}_{2}$ in Figure 2. While computing $\nabla_{blue}$, it receives the yellow mini-batch from the Mini Batch generator, 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}$, Model${}_{2}$ receives $\nabla_{green}$ to apply, which it does promptly after finishing $\nabla_{yellow}$. Note that when Model${}_{3}$ computed $\nabla_{green}$ and Model${}_{1}$ computed $\nabla_{red}$, they have not applied a single gradient to their local models at that time. Thus, $|G_{1}|=|G_{3}|=0$. However, before applying $\nabla_{green}$, $G_{2}=\{\nabla_{blue},\nabla_{yellow}\}$ with $|G_{2}|=2$ and $\mathit{staleness}_{3,2}(\nabla_{green})=2$. Along the same lines, before applying $\nabla_{red}$, $|G_{2}|=3$ and $\mathit{staleness}_{1,2}(\nabla_{red})=3$.

3.2. 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 that most SGD proofs of convergence (Zhang etal., 2020; Sun etal., 2017; Jiang etal., 2017) do not cover. Therefore, we next discuss under which circ*mstances TensAIR is guaranteed to converge.

First, we consider that training is performed between significant concept drifts. Therefore, we assume that the data distribution between two subsequent concept drifts does 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 (Goodfellow etal., 2016), each of the local model instances will eventually converge.

Second, considering TensAIR’s decentralized and asynchronous SGD (DASGD), model updates can be staled. Nevertheless, as proven by Tosi and Theobald (Tosi and Theobald, 2023), the model will converge in this setting in up to $\mathcal{O}(\frac{\sigma}{\epsilon^{2}})+\mathcal{O}(\frac{QS_{avg}}{\epsilon^$ iterations to an $\epsilon$-small error, considering $S_{avg}$ as the average staleness observed during training and $Q$ a constant that bounds the gradients size. If bounded gradients are not assumed, DASGD converges in $\mathcal{O}(\frac{\sigma}{\epsilon^{2}})+\mathcal{O}(\frac{\sqrt{\hat{S}_{avg}$ iterations, with $\hat{S}_{max}$ and $\hat{S}_{avg}$ representing the maximum and average staleness, calculated using an additional recursive factor.

3.3. Implementation

TensAIR was implemented on top of the Asynchronous Iterative Routing (AIR) (Venugopal etal., 2020, 2022) dataflow engine. AIR is a native stream-processing engine that processes complex dataflows in an asynchronous and decentralized manner. TensAIR dataflow operators extend a basic Vertex superclass in AIR. 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. This is crucial to guarantee that worker nodes do not stay idle while waiting to send or receive messages during training. The number of instances of each Vertex subclass and the number of input data streams can be configured beforehand, as seen in Figure 1.

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 GitHub repository¹¹1https://github.com/maurodlt/TensAIR.

4.1. Convergence Analysis

We first explored TensAIR’s ability to converge by determining if and how DASGD 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 2(b) & 2(d)). We additionally explored the models convergence when trained with distributed CPU nodes (Figures 2(a) & 2(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.

The extremely low variance among all loss curves shown in Figures 2(a) and 2(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 2(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 $b$ is similar to training 1 model with mini-batches of size $N\cdot b$. To mitigate this issue, Horovod increases the learning rate $\alpha$ by the number of ranks used to distribute the model (hor, 2019), i.e., $\alpha_{new}=\alpha\cdot N$. Accordingly, in Figure 2(d), we again do not see any degradation of the loss when distributing CIFAR-10 because we use a maximum of 4 GPUs.

4.2. 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, differently than the gradient calculation, the gradient application is not distributed and, with smaller mini-batches, more gradients are applied per epoch. Thus, models with expensive gradient computations will have a better scale-out performance. Nevertheless, when gradient calculation is not the bottleneck of the dataflow, one can reduce the computational impact of the gradients application and the network impact of their broadcasts by simply increasing maxBuffer. For instance, by increasing maxBuffer in $n$ times, the network complexity and the computational impact of the gradients applications are also expected to be reduce in $n$ times.

4.3. 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 (kaf, 2022). 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 (dl-, 2022) even after various attempts on our HPC setup. We therefore report the direct deployment of our Word2Vec and CIFAR-10 use-cases (Figures 4(a) & 4(b)) 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 (Sergeev and DelBalso, 2018), which is however not a streaming engine by default.

In Figures 4(a) and 4(b), 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 was designed to train ANN models from high throughput data streams in real-time. Thus, the transfer of training data overhead is mitigated by the asynchronous protocol adopted and the training is speed-up by DASGD. 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. Note that, while using more than 3 GPUs simultaneously, TensAIR did not achieve better sustainable throughput in the W2V usecase due to the relatively low complexity of the gradient calculations. In this scenario, the training bottleneck, typically associated with gradient calculation, shifted to the gradient application when using more than 3 GPUs, as the former is not distributed. Nevertheless, this issue can be mitigated by simply increasing the variable $maxBuffer$ (as explained in Section 4.2). This adjustment, delays the communication among distributed models while reducing the locally applied gradients by a factor of $maxBuffer$.

4.4. 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 (Go etal., 2009). Then, we trained an LSTM model (ten, 2022) 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.

As shown in Table 1, 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 6) 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.

Figure6 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 (Model${}^{SA}$). However, it does not rely on pre-defined word embeddings. The dataflow constantly improves its embeddings on a second Word2Vec (W2V) ANN model (Model${}^{W2V}$), 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 Model${}^{SA}$, since only Model${}^{W2V}$ is re-trained in a self-supervised manner by generating mini-batches of word pairs $(x,y)$ directly from the input tweets.

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 Model${}^{W2V}$ and Model${}^{SA}$. 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. Model${}^{W2V}$ 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 Model${}^{W2V}$ instances and sent to the pre-trained Model${}^{SA}$ which then predicts the tweets’ sentiments.