#  Persia: An Open, Hybrid System Scaling Deep Learning-based Recommenders up
to 100 Trillion Parameters

Xiangru Lian1, Binhang Yuan3, Xuefeng Zhu2, Yulong Wang2, Yongjun He3,
Honghuan Wu2, Lei Sun2, Haodong Lyu2, Chengjun Liu2, Xing Dong2, Yiqiao Liao2,
Mingnan Luo2, Congfei Zhang2, Jingru Xie2, Haonan Li2, Lei Chen2, Renjie
Huang2, Jianying Lin2, Chengchun Shu2, Xuezhong Qiu2, Zhishan Liu2, Dongying
Kong2, Lei Yuan2, Hai Yu2, Sen Yang2, Ce Zhang3, Ji Liu1 1Kwai Inc. USA;
2Kuaishou Technology, China; 3ETH Zürich, Switzerland;
<EMAIL_ADDRESS>2.kuaishou.com; 3.inf.ethz.ch

(2022)

###### Abstract.

Deep learning based models have dominated the current landscape of production
recommender systems. Furthermore, recent years have witnessed an exponential
growth of the model scale—from Google’s 2016 model with 1 billion parameters
to the latest Facebook’s model with 12 trillion parameters. Significant
quality boost has come with each jump of the model capacity, which makes us
believe the era of 100 trillion parameters is around the corner. However, the
training of such models is challenging even within industrial scale data
centers. This difficulty is inherited from the staggering heterogeneity of the
training computation—the model’s embedding layer could include more than
$99.99\%$ of the total model size, which is extremely memory-intensive; while
the rest neural network is increasingly computation-intensive. To support the
training of such huge models, an efficient distributed training system is in
urgent need. In this paper, we resolve this challenge by careful co-design of
both the optimization algorithm and the distributed system architecture.
Specifically, in order to ensure both the training efficiency and the training
accuracy, we design a novel hybrid training algorithm, where the embedding
layer and the dense neural network are handled by different synchronization
mechanisms; then we build a system called Persia (short for parallel
recommendation training system with hybrid acceleration) to support this
hybrid training algorithm. Both theoretical demonstrations and empirical
studies up to 100 trillion parameters have been conducted to justified the
system design and implementation of Persia. We make Persia publicly available
(at https://github.com/PersiaML/Persia) so that anyone would be able to easily
train a recommender model at the scale of 100 trillion parameters.

††copyright: acmcopyright††journalyear: 2022††doi:
10.1145/1122445.1122456††conference: SIGMOD ’22: International Conference on
Management of Data; June 12–17, 2022; Philadelphia, USA

##  1\. Introduction

A recommender system is an important component of Internet services today.
Tasks such as click-through rate (CTR) and buy-through rate (BTR) predictions
are widely adopted in industrial applications, influencing the ad revenues at
billions of dollar level for search engines such as Google, Bing and Baidu (Su
et al., 2017). Moreover, $80\%$ of movies watched on Netflix (Gomez-Uribe and
Hunt, 2015) and $60\%$ of videos clicked on YouTube (Davidson et al., 2010)
are driven by automatic recommendations; over $40\%$ of user engagement on
Pinterest are powered by its Related Pins recommendation module (Liu et al.,
2017); over half of the Instagram community has visited recommendation based
Instagram Explore to discover new content relevant to their interests (Acun et
al., 2020); up to $35\%$ of Amazon’s revenue is driven by recommender systems
(Xie et al., 2018; Chang et al., 2021). At Kwai, we also observe that
recommendation plays an important role for video sharing—more than 300 million
of daily active users explore videos selected by recommender systems from
billions of candidates.

Figure 1.  Model sizes of different recommender systems, among which only XDL
and AIBox (via PaddlePaddle) are open-source. Persia is an open-source
training system for deep learning-based recommender systems, which scales up
models to the scale of 100 trillion parameters.

Racing towards 100 trillion parameters. The continuing advancement of modern
recommender models is often driven by the ever increasing model sizes—from
Google’s 2016 model with 1 billion parameters (Covington et al., 2016) to
Facebook’s latest model (2022) with 12 trillion parameters (Mudigere et al.,
2021) (See Figure 1). Every jump in the model capacity has been bringing in
significantly improvement on quality, and the era of 100 trillion parameters
is just around the corner.

Interestingly, the increasing parameter comes mostly from the embedding layer
which maps each entrance of an ID type feature (such as an user ID (Li et al.,
2018b; Tang and Wang, 2018) and a session ID (Tan et al., 2016; Twardowski,
2016; Tuan and Phuong, 2017)) into a fixed length low-dimensional embedding
vector. Consider the billion scale of entrances for the ID type features in a
production recommender system (e.g., (Eksombatchai et al., 2018; Wang et al.,
2018a)) and the wide utilization of feature crosses (Cheng et al., 2016), the
embedding layer usually domains the parameter space, which makes this
component extremely memory-intensive. On the other hand, these low-dimensional
embedding vectors are concatenated with diversified Non-ID type features
(e.g., image (Wang et al., 2017b; Wen et al., 2018), audio (Van Den Oord et
al., 2013; Wang and Wang, 2014), video (Chen et al., 2017; Lee et al., 2018),
social network (Hsieh et al., 2016; Deng et al., 2016), etc.) to feed a group
of increasingly sophisticated neural networks (e.g., convolution, LSTM, multi-
head attention) for prediction(s) (Covington et al., 2016; Zhou et al., 2018b,
2019; Zhang et al., 2020; Naumov et al., 2020; Zhao et al., 2020a).
Furthermore, in practice, multiple objectives can also be combined and
optimized simultaneously for multiple tasks (Wang et al., 2016, 2018c; Lu et
al., 2018; Zhao et al., 2019a; Li et al., 2020). These mechanisms make the
rest neural network increasingly computation-intensive.

Challenges. This paper is motivated by the challenges that we were facing when
applying existing systems, e.g., XDL from Alibaba (Jiang et al., 2019) and
PaddlePaddle from Baidu (pad, [n.d.]), to the training of models at the 100
trillion parameter scale. This is a challenging task—as illustrated in Figure
1, most open source systems were only designed for a scale that is at least
one order of magnitude smaller; even the largest proprietary systems such as
the one reported by Facebook (Mudigere et al., 2021) just two month ago
(September 2021) are still $8.3\times$ smaller. The goal of this paper is to
enable, to our best knowledge, the first open source system that is able to
scale in this regime.

As 100 trillion parameters require at least 200TB to simply store the model
(even in fp16), a distributed training system at this scale often consists of
hundreds of machines. As so, the communication among workers is often a system
bottleneck. While state-of-the-art systems (Jiang et al., 2019; Raman et al.,
2019; Zhao et al., 2019b; Pan et al., 2020; Zhao et al., 2020b; Naumov et al.,
2020; Krishna and Krishna, 2020; Wilkening et al., 2021) often employ a
carefully designed heterogeneous architecture (e.g., CPU/GPU, DRAM/SSD) to
accommodate the heterogeneity in a recommender model (as illustrated in Figure
2), all of them are using a homogeneous training algorithm (either synchronous
or asynchronous stochastic gradient based algorithms), for the model as a
whole. At the scale of 100 trillion parameters and hundreds of workers, this
homogeneity in the training algorithm starts to become a significant problem:

(1) Synchronous algorithms always use the up-to-date gradient to update the
model to ensure the model accuracy. However, the overhead of communications
for synchronous algorithms starts to become too expensive to scale out the
training procedure, causing inefficiency in running time.

(2) While asynchronous algorithm have better hardware efficiency, it often
leads to a “significant” loss in model accuracy at this scale—for production
recommender systems (e.g., Baidu’s search engine (Zhao et al., 2020b)). Recall
that even 0.1% drop of accuracy would lead to a noticeable loss in
revenue—this is also consistent with our observation at Kwai.

Motivated by these challenges, we ask:

> Q1. Can we design an algorithm that can take benefits from both synchronous
> and asynchronous updates avoiding their disadvantages, to further scale up a
> recommender system with 100 trillion parameters?

> Q2. How can we design, optimize, and implement a system to efficiently
> support such an algorithm?

Persia. In this paper, we describe Persia, an open source distributed training
system developed at Kwai to support models at the scale of 100 trillion
parameters. As we will show in Section 6, this is only possible given its
significant speed-up over state-of-the-art systems including both XDL and
PaddlePaddle, achieved by the careful co-design of both the training algorithm
and the training system. Open Source and Reproducibility: We also believe that
the ability to train such a large model should be made easily and widely
available to everyone instead of just being locked in the hands of a handful
of largest companies. We make Persia open source so that everyone who has
access to cloud computing service, e.g., Google cloud platform (where we
perform Persia’s capacity test) can easily setup a distributed training system
to reproduce our 100-trillion-parameter model test and to train their own
models at this scale111One can follow our detailed configuration of Persia
over Google cloud platform at this link:
https://github.com/PersiaML/tutorials/blob/main/src/kubernetes-
integration/index.md)..

Figure 2. An example of a recommender models with $100+$ trillions of
parameter in the embedding layer and $50+$ TFLOP computation in the neural
network.

Contributions. Persia is enabled by a set of technical contributions. The core
technical hypothesis of Persia is that, by using a hybrid and heterogeneous
training algorithm, together with a heterogeneous system architecture design,
we can further improve the performance of training recommender systems over
state-of-the-arts.

Our first contribution is a natural, but novel hybrid training algorithm to
tackle the embedding layer and dense neural network modules differently—the
embedding layer is trained in an asynchronous fashion to improve the
throughput of training samples, while the rest neural network is trained in a
synchronous fashion to preserve the statistical efficiency. We also provide a
rigorous theoretical analysis on its convergence behavior and further connect
the characteristics of a recommender model to its convergence to justify its
effectiveness.

This hybrid algorithm requires us to revisit some decisions with respect to
the system architecture and optimizations, to unleash its full potential. Our
second contribution is on the system side. We design a distributed system to
manage the hybrid computation resources (CPUs and GPUs) to optimize the co-
existence of asynchronicity and synchronicity in the training algorithm. We
further implement a wide range of system optimizations for memory management
and communication optimization—such system optimizations are the key to fully
unleash the potential of our hybrid training algorithm. We also develop a
fault tolerance strategy to handle various potential failures during the
training procedure.

Last but not least, we evaluate Persia using both publicly available benchmark
tasks and real-world tasks at Kwai. We show that Persia scales out effectively
and leads to up to $7.12\times$ speedup compared to alternative state-of-the-
art approaches (Jiang et al., 2019; pad, [n.d.]). We further conduct larger-
scale scalability experiments up to 100 trillion parameters on public clouds
to ensure public reproducibility.

Overview. The rest of the paper is organized as follows. We first provide some
preliminaries for deep learning recommender systems in Section 2. We introduce
our hybrid training algorithm in Section 3 and discuss Persia system design
and implementation in Section 4. We show the theoretical analysis of the
hybrid algorithm in Section 5, present the experimental results in Section 6,
summarize related work in Section 7, and conclude in Section 8.

##  2\. Preliminaries

This section introduces the basic principle of a typical deep learning based
recommendation system (e.g., DLRM (Mudigere et al., 2021)) illustrated in
Figure 2. We first formalize the distributed recommender training problem; and
then give an anatomy of the existing architectures.

###  2.1. Problem Formalization

A typical recommender system takes the training sample in the form:

$\displaystyle\left[\mathbf{x}^{\mathrm{ID}}_{\xi},\mathbf{x}^{\mathrm{NID}}_{\xi},\mathbf{y}_{\xi}\right]$

where $\xi$ denotes the index of the sample in the whole dataset.
$\mathbf{x}^{\mathrm{ID}}_{\xi}:=\\{x_{1;\xi},x_{2;\xi},\ldots\\}$ is the
collection of ID type features in the sample,
$\left|\mathbf{x}^{\mathrm{ID}}_{\xi}\right|$ denotes the number of IDs,
$\mathbf{x}^{\mathrm{NID}}_{\xi}$ denotes the Non-ID type features, and
$\mathbf{y}_{\xi}$ denotes the label. The ID type feature is the sparse
encoding of large-scale categorical information. For example, one may use a
group of unique integers to record the microvideos (e.g., noted as
$\langle\texttt{VideoIDs}\rangle$) that have been viewed by a user; similar ID
type features may include location ($\langle\texttt{LocIDs}\rangle$), relevant
topics ($\langle\texttt{TopicIDs}\rangle$), followed video bloggers
($\langle\texttt{BloggerIDs}\rangle$), etc. In our formalization,
$\mathbf{x}^{\mathrm{ID}}$ is the collection of all ID type features—for the
above example, it can be considered as:

$\displaystyle\mathbf{x}^{\mathrm{ID}}:=\left[\langle\texttt{VideoIDs}\rangle,\langle\texttt{LocIDs}\rangle,\langle\texttt{TopicIDs}\rangle,\langle\texttt{BloggerIDs}\rangle,...\right]$

The Non-ID type feature $\mathbf{x}^{\mathrm{NID}}$ can include various visual
or audio features. And the label $\mathbf{y}$ may include one or multiple
value(s) corresponding to one or multiple recommendation task(s).

The parameter $\mathbf{w}$ of the recommender system usually has two
components:

$\displaystyle\mathbf{w}:=\left[\mathbf{w}^{\mathrm{emb}},\mathbf{w}^{\mathrm{nn}}\right]\in\mathbb{R}^{N^{\mathrm{emb}}+N^{\mathrm{nn}}}$

where $\mathbf{w}^{\mathrm{emb}}\in\mathbb{R}^{N^{\mathrm{emb}}}$ is the
parameter of the embedding layer and
$\mathbf{w}^{\mathrm{nn}}\in\mathbb{R}^{N^{\mathrm{nn}}}$ is the parameter of
the rest dense neural network. We use:
$\mathrm{lookup}_{\mathbf{w}^{\mathrm{emb}}}\left(\mathbf{x}^{\mathrm{ID}}\right)$
to denote the concatenation of all embedding vectors that has correspondence
in $\mathbf{x}^{\mathrm{ID}}$;
$\mathrm{NN}_{\mathbf{w}^{\mathrm{nn}}}\left(\cdot\right)$ to denote a
function parameterized by $\mathbf{w}^{\mathrm{nn}}$ implemented by a deep
neural network that takes the looked up embeddings and Non-ID features as
input and generates the prediction. In the remaining part of this paper, “NN”
is short for “neural network”. The recommender system predicts one or multiple
values $\hat{\mathbf{y}}$ by:

$\displaystyle\hat{\mathbf{y}}_{\xi}=\mathrm{NN}_{\mathbf{w}^{\mathrm{nn}}}\left(\mathrm{lookup}_{\mathbf{w}^{\mathrm{emb}}}\left(\mathbf{x}^{\mathrm{ID}}_{\xi}\right),\mathbf{x}^{\mathrm{NID}}_{\xi}\right)$

It is worth mentioning that while the $\mathbf{w}^{\mathrm{nn}}$ involved
computation can be $10^{7}\times$ more than the $\mathbf{w}^{\mathrm{emb}}$
involved computation, the size of $\mathbf{w}^{\mathrm{emb}}$ can be
$10^{7}\times$ larger than that of $\mathbf{w}^{\mathrm{nn}}$, especially when
$\mathbf{w}^{\mathrm{emb}}$ contains many cross features. The imbalance
between model scale and model computation intensity is one of the fundamental
reasons why the recommender system needs an elaborate training system other
than a general purpose deep learning system. Formally, the training system
essentially solves the following optimization

(1) $\displaystyle\min_{\mathbf{w}}\quad f\left(\mathbf{w}\right):=$
$\displaystyle\mathbb{E}_{\xi}\left[F\left(\mathbf{w};\xi\right)\right].$

if we use $\mathcal{L}$ to denote some loss function over the prediction and
the true label(s) $\mathbf{y}$, $F$ can be materialized as:

$\displaystyle
F(\mathbf{w};\xi):=\mathcal{L}\left(\mathrm{NN}_{\mathbf{w}^{\mathrm{nn}}}\left(\mathrm{lookup}_{\mathbf{w}^{\mathrm{emb}}}\left(\mathbf{x}^{\mathrm{ID}}_{\xi}\right),\mathbf{x}^{\mathrm{NID}}_{\xi}\right),\mathbf{y}_{\xi}\right).$

Lastly, we use

$\displaystyle{F_{\xi}^{\mathrm{emb}}}^{\prime}:=\nabla_{\mathbf{w}^{\mathrm{emb}}}F\left(\mathbf{w}^{\mathrm{emb}},\mathbf{w}^{\mathrm{nn}};\xi\right)$
$\displaystyle{F_{\xi}^{\mathrm{nn}}}^{\prime}:=\nabla_{\mathbf{w}^{\mathrm{nn}}}F\left(\mathbf{w}^{\mathrm{emb}},\mathbf{w}^{\mathrm{nn}};\xi\right)$

to denote the gradients of $\mathbf{w}^{\mathrm{emb}}$ and
$\mathbf{w}^{\mathrm{nn}}$ respectively.

###  2.2. Anatomy of Existing Architectures

Existing distributed systems for deep learning based recommender models are
usually built on top of the parameter server (PS) framework (Li et al., 2014),
where one can add elastic distributed storage to hold the increasingly large
amount of parameters of the embedding layer. On the other hand, the
computation workload does not scale linearly with the increasing parameter
scale of the embedding layer—in fact, with an efficient implementation, a
lookup operation over a larger embedding table would introduce almost no
additional computations. Thus, a fundamental goal of these systems is to
resolve the inconsistency between the computation-intensive neural network and
the memory-intensive embedding layer.

One natural solution is to deploy the training task over a hybrid
infrastructure as an extension of the original PS framework: the large number
of embedding parameters can be sharded across multiple CPU PS nodes for
storage and update, whilst the intensive FLOP computation can be assigned to
GPU nodes. Thus, the optimization in these systems focuses on avoiding
frequent communications of the memory-intensive embedding layer. Towards this
end, different designs have been proposed. For example, Alibaba proposes XDL
(Jiang et al., 2019) that moves the computation of the embedding layer from
the GPU worker to the CPU PS node so that only the embedding outputs (instead
of all parameters) are transformed to the GPU workers along with the neural
network parameters; Baidu designs a hierarchical PS framework (Liu and Zhang,
2020) using a hierarchical storage to cache the frequently used parameter
close to the GPU workers—only the infrequently used parameters would trigger
the slow communication from SSD to GPU.

Our Motivation. Existing systems such as XDL and PaddlePaddle update the model
in either the pure synchronous fashion or the pure asynchronous fashion. Both
generally perform well on training small models. However, when training a
large-scale model, the synchronous updating suffers from low hardware
efficiency while the asynchronous updating usually incurs low accuracy (recall
that a tiny loss in accuracy often means a huge loss in revenue in the
recommender system). Therefore, it motivates us to design a (sync-async)
hybrid algorithm and an efficient system that is able to take benefits from
both synchronous and asynchronous updates but avoid their disadvantages.

Figure 3. Left: deep learning based recommender model training workflow over a
heterogeneous cluster. Right: Gantt charts to compare fully synchronous, fully
asynchronous, raw hybrid and optimized hybrid modes of distributed training of
the deep learning recommender model.

##  3\. Hybrid Training Algorithm

This section starts with explaining the intuition of the proposed hybrid
algorithm, followed by the detailed algorithm description.

###  3.1. Why Hybrid Algorithm?

Training a recommendation mode requires the following essential steps in each
iteration as shown in Figure 3-left:

  * •

Preparation of embedding for the training sample(s);

  * •

Forward propagation of the neural network;

  * •

Backward propagation of the neural network;

  * •

Synchronization of the parameters of the neural network;

  * •

Update of embedding based on the corresponding gradients.

One illustration of the training workflow is shown on in Figure 3-left. Note
that a homogeneous view of the embedding layer and NN would set obstacles for
either hardware or statistical efficiency. For example, consider the fully
synchronous mode that we illustrate in Figure 3-right, the above five steps
have to happen sequentially—it would be hard to speed up the training
procedure based on this mechanism with low hardware efficiency.

Asynchronous distributed training. It is important to consider appropriate
system relaxations for stochastic gradient-based optimizations222This category
of optimization includes SGD, Adam(Kingma and Ba, 2014), etc. We slightly
abuse the term “SGD” to generally refer such optimizations for the rest of the
paper. deployed in a hybrid distributed runtime. There are two key
observations from the ML community about asynchronous distributed training of
neural networks:

  * •

Asynchronous updating is efficient with sparse access. When individual
updating (e.g., SGD iteration) only modifies a small portion of the model’s
parameters, overwrites are rare and introduce barely no bias into the
computation when they do occur (Niu et al., 2011; Lian et al., 2016).

  * •

Staleness limits the scalability and convergence of asynchronous SGD. On the
other hand, when the updates are heavily overlapped (e.g, dense neural
networks), the bias introduced by the discrepancy would limit SGD’s
scalability and convergence (Lian et al., 2015; Chen et al., 2016).

Without considering the statistical efficiency, the fully asynchronous mode
(second row in Figure 3-right) would provide the optimal hardware efficiency
for distributed recommender model training, where the time of preparing
embeddings, synchronizing dense parameters, and updating embedding parameters
can be hidden within the computation time for forward and backward
propagations of the dense neural network. Unfortunately, as we will illustrate
in Section 6.2, the asynchronicity would hurt the statistical efficiency and
diminish the generalization performance, which is not acceptable for
production recommender models (Zhao et al., 2020b).

Hybrid training algorithm. Based on these observations, one may consider to
design different mechanisms for the embedding and dense neural network
separately to optimize the training efficiency. For this purpose, we introduce
a natural, but novel sync-async hybrid algorithm, where the embedding module
trains in an asynchronous fashion while the dense neural network is updated
synchronously. Considering the inherited heterogeneity of deep learning based
recommender systems, this design would exploit the strength whilst avoid the
weaknesses of asynchronous SGD from the algorithmic perspective. Briefly, as
we illustrate in the third and forth rows in Figure 3-right, a naive hybrid
mode is able to hide the steps of preparing embeddings, and updating embedding
parameters within the synchronous training of the dense neural network;
further, with some advanced system optimizations (e.g., overlapping
computation and communication listed in Section 4.2.3), the synchronization of
the dense parameters would also be able to mostly hidden within the backward
propagation of the dense neural network. Thus, the hybrid algorithm would
achieve almost similar hardware efficiency as the fully asynchronous mode
without sacrificing the statistical efficiency.

###  3.2. Asynchronous updating for $\mathbf{w}^{\mathrm{emb}}$

To discuss the asynchronous algorithm to update embedding parameter
$\mathbf{w}^{\mathrm{emb}}$, given training sample indexed by $\xi$, we
introduce the following operations:

  * •

get$\left({\mathbf{x}_{\xi}^{\mathrm{ID}}}\right)$: fetch the subset of
parameters to generate the embedding of the ID type feature
${\mathbf{x}_{\xi}^{\mathrm{ID}}}$—we use the notation
$\mathbf{w}^{\mathrm{emb}}_{\mathrm{\xi}}$ to represent this subset of
parameters activated by the sample.

  * •

put$\left({\mathbf{x}_{\xi}^{\mathrm{ID}}},{{F^{\mathrm{emb}}_{\xi}}}^{\prime}\right)$:
communicate the gradient ${{F^{\mathrm{emb}}_{\xi}}}^{\prime}$ w.r.t
${\mathbf{x}_{\xi}^{\mathrm{ID}}}$ to the storage of parameter
$\mathbf{w}^{\mathrm{emb}}$.

Algorithm 1 shows the asynchronous updating algorithm for the embedding layer
$\mathbf{w}^{\mathrm{emb}}$. Both forward computation task and backward
computation task will be executed without any lock for synchronization: the
forward computation task takes training samples, retrieves the embedding
vectors, and sends the corresponding embedding vectors to the proceeding
compute units (which handle the NN SGD computation); the backward computation
task will receive the gradients of the embedding vectors, and sends it to the
parameter storage for parameter updates.

Context: Forward task for embedding layer.

Input: Embedding layer $lookup\left(\cdot\right)$, training data
$\mathcal{D}$.

while Not converge do

/* Without any lock: */

Select a sample from the training set:
$\mathbf{x}^{\mathrm{ID}}_{\xi}\sim\mathcal{D}$;

Get embedding vector(s) w.r.t the sample:
$\mathbf{w}^{\mathrm{emb}}_{\mathrm{\xi}}\leftarrow$
get$\left({\mathbf{x}_{\xi}^{\mathrm{ID}}}\right)$;

Send the embedding vector(s) to proceeding units:
$\mathbf{w}^{\mathrm{emb}}_{\mathrm{\xi}}\downarrow$.

end while

Context: Backward task for embedding layer.

Input: Embedding layer $lookup\left(\cdot\right)$, optimizer
$\Omega^{\mathrm{emb}}$.

while Activation’s gradients keeps arriving do

/* Without any lock: */

Receive gradient of embedding: ${{F^{\mathrm{emb}}_{\xi}}}^{\prime}$;

Send gradients to parameter storage:
put$\left({\mathbf{x}_{\xi}^{\mathrm{ID}}},{{F^{\mathrm{emb}}_{\xi}}}^{\prime}\right)$;

Update embedding parameter:
$\mathbf{w}^{\mathrm{emb}}_{\mathrm{t+1}}\leftarrow\mathbf{w}^{\mathrm{emb}}_{\mathrm{t}}-\Omega^{\mathrm{emb}}\left({F_{\xi}^{\mathrm{emb}}}^{\prime}\left\\{\xi\right\\}\right)$.

end while

Algorithm 1 Asynchronous updating algorithm for $\mathbf{w}^{\mathrm{emb}}$.

Context: A task for dense module (indexed by $k$, $k=1,2,...,K$).

Input: Neural network
$\mathrm{NN}_{\mathbf{w}^{\mathrm{nn}}}\left(\cdot\right)$, embeddings
$\mathbf{w}^{\mathrm{emb}}_{\mathrm{\left(\cdot\right)}}$, optimizer
$\Omega^{\mathrm{nn}}$.

while Not converge do

Randomly select $b$ buffered embeddings:
$\mathbf{w}^{\mathrm{emb}}_{\mathrm{\xi_{1}}},...,\mathbf{w}^{\mathrm{emb}}_{\mathrm{\xi_{b}}}\sim\mathbf{w}^{\mathrm{emb}}_{\mathrm{\left(\cdot\right)}}$;

Forward of batch B:
$F_{B}=\displaystyle\sum\limits_{i=1}^{b}\mathcal{L}\left(\mathrm{NN}_{\mathbf{w}^{\mathrm{nn}}}\left(\mathbf{w}^{\mathrm{emb}}_{\mathrm{\xi_{i}}},\mathbf{x}^{\mathrm{NID}}_{\xi_{i}}\right),\mathbf{y}_{\xi_{i}}\right)$;

/* With locks as synchronization barrier: */

Backward of batch B: compute gradient ${{F^{\mathrm{nn}}_{B}}}^{\prime}$, and
${{F^{\mathrm{emb}}_{\xi_{i}}}}^{\prime}$ ;

Sync gradients with optimization:
${{F^{\mathrm{nn}}_{B}}}^{\prime}\leftarrow\frac{1}{K}\displaystyle\sum\limits_{k=1}^{K}\left[{{F^{\mathrm{nn}}_{B}}}^{\prime}\right]_{k}$

Update dense parameter:
$\mathbf{w}^{\mathrm{nn}}_{\mathrm{t+1}}\leftarrow\mathbf{w}^{\mathrm{nn}}_{\mathrm{t}}-\Omega^{\mathrm{nn}}\left({{F^{\mathrm{nn}}_{B}}}^{\prime}\right)$.

Send back activation’s gradients:
$\left[{{F^{\mathrm{emb}}_{\xi_{1}}}}^{\prime},{{F^{\mathrm{emb}}_{\xi_{2}}}}^{\prime}...{{F^{\mathrm{emb}}_{\xi_{b}}}}^{\prime}\right]\uparrow$

end while

Algorithm 2 Synchronous updating algorithm for $\mathbf{w}^{\mathrm{nn}}$.

###  3.3. Synchronous updating for $\mathbf{w}^{\mathrm{nn}}$

Our training algorithm for the rest neural network looks similar to the
standard distributed training of deep neural networks—the key difference is
that our algorithm concatenates the embedding activations as part of input.
The algorithm is illustrated in Algorithm 2. Notice that we adopt a mini-batch
based SGD algorithm in contrast to the sample-based SGD in the asynchronous
counterpart.

Since the hybrid algorithm is quite different from what has been assumed by
existing systems (Li et al., 2014; Sergeev and Del Balso, 2018; Jiang et al.,
2019; Liu and Zhang, 2020; pad, [n.d.]). To fully unleash its potential, we
have to carefully design the system and optimize its performance—we introduce
the system design and implementation of Persia in Section 4. On the other
hand, it is also important to understand the statistical efficiency of the
proposed hybrid training algorithm, thus we provide a theoretical analysis
about the convergence guarantee of the hybrid algorithm in Section 5.

##  4\. System Design and Implementation

In this section, we first introduce the design of Persia to support the hybrid
algorithm; then we discuss a wide range of implementations to optimize the
computation and communication utilization.

###  4.1. System Design

The system design includes two main fundamental aspects: i) the placement of
the training workflow over a heterogeneous cluster, and ii) the corresponding
training procedure over the hybrid infrastructure. To enlighten the system
implementation, we also list the implementation goals of the Persia system
here.

Workflow placement over heterogeneous cluster. To support the distributed
training of deep learning based recommender model, a straightforward
utilization of the PS paradigm (provided by general purpose deep learning
frameworks, e.g., TensorFlow (Abadi et al., 2016)) would place the storage and
update of both embedding and NN parameters in a group of PS nodes (i.e., CPU
machines) and the computation of forward and backward propagations in a group
of worker nodes (i.e., GPU machines). However, this would be far from
efficient and even impossible for deployment. For example, such recommender
model would easily exceed the GPU RAM; and the uniform view of embedding and
NN modules would introduce a large amount of unnecessary network traffic.

Some optimized PS architectures are proposed to optimize the training of deep
learning recommender models by rearrange the functionalities in the PS
paradigm. For example, XDL (Jiang et al., 2019) designs the advanced model
server to extend the original functionality of PS node to manage the learning
(forward and backward propagations) of the embedding module. The hierarchical
PS architecture proposed by Baidu (Zhao et al., 2020b) adopts a colocated PS
framework with a sophisticated caching schema to reduce the communication
overhead.

New challenges for the layout of 100-trillion-parameter models. In order to
support the recommender models with one or two magnitude larger models, Persia
should provide efficient autoscaling. Thus, we introduce the following
modules, where each module can be dynamically scaled for different model
scales and desired training throughput:

  * •

A data loader that fetches training data from distributed storages such as
Hadoop, Kafka, etc;

  * •

An embedding parameter server (embedding PS for short) that manages the
storage and update of the parameters in the embedding layer
$\mathbf{w}^{\mathrm{emb}}$;

  * •

A group of embedding workers that runs Algorithm 1 for getting the embedding
parameters from the embedding PS; aggregating embedding vectors (potentially)
and putting embedding gradients back to embedding PS;

  * •

A group of NN workers that runs the forward-/backward- propagation of the
neural network $\mathrm{NN}_{\mathbf{w}^{\mathrm{nn}}}\left(\cdot\right)$.

Considering the heterogeneity between the embedding and NN modules, Persia
adopts different communication paradigms for training process: i) the PS
paradigm between embedding PS and embedding workers (running on CPU nodes) to
manage the training of the embedding layer, while ii) the AllReduce paradigm
among NN workers (running on GPU nodes) for the NN.

Figure 4. The architecture of Persia. Persia includes a data loader module, an
embedding PS module, a group of embedding workers over CPU nodes, and a group
of NN workers over GPU instances.

Distributed training procedure. Logically, the training procedure is conducted
by Persia in a data dispatching based paradigm as below (see Figure 4):

  1. (1)

The data loader will dispatch the ID type feature
${\mathbf{x}_{\left(\cdot\right)}^{\mathrm{ID}}}$ to an embedding worker—the
embedding worker will generate a unique sample ID333Note that the unique ID
$\xi$ will be used to locate the embedding worker that generates this ID—this
could simply be implemented by using the first byte to encode the rank of this
embedding worker. $\xi$ for this sample, buffer this sample ID with the ID
type feature ${\mathbf{x}_{\xi}^{\mathrm{ID}}}$ locally, and return this ID
$\xi$ back the data loader; the data loader will associate this sample’s Non-
ID type features and labels with this unique ID.

  2. (2)

Next, the data loader will dispatch the Non-ID type feature and label(s)
$\left({\mathbf{x}_{\xi}^{\mathrm{NID}}},\mathbf{y}_{\xi}\right)$ to a NN
worker.

  3. (3)

Once a NN worker receives this incomplete training sample, it will issue a
request to pull the ID type features’ ($\mathbf{x}^{\mathrm{ID}}_{\xi}$)
embedding $\mathbf{w}^{\mathrm{emb}}_{\mathrm{\xi}}$ from some embedding
worker according to the sample ID $\xi$—this would trigger the forward
propagation in Algorithm 1, where the embedding worker will use the buffered
ID type feature ${\mathbf{x}_{\xi}^{\mathrm{ID}}}$ to get the corresponding
$\mathbf{w}^{\mathrm{emb}}_{\mathrm{\xi}}$ from the embedding PS.

  4. (4)

Then the embedding worker performs some potential aggregation of original
embedding vectors. When this computation finishes, the aggregated embedding
vector $\mathbf{w}^{\mathrm{emb}}_{\mathrm{\xi}}$ will be transmitted to the
NN worker that issues the pull request.

  5. (5)

Once the NN worker gets a group of complete inputs for the dense module, it
will create a mini-batch and conduct the training computation of the NN
according to Algorithm 2. Note that the parameter of the NN always locates in
the device RAM of the NN worker, where the NN workers synchronize the
gradients by the AllReduce Paradigm.

  6. (6)

When the iteration of Algorithm 2 is finished, the NN worker will send the
gradients of the embedding (${{F^{\mathrm{emb}}_{\xi}}}^{\prime}$) back to the
embedding worker (also along with the sample ID $\xi$).

  7. (7)

The embedding worker will query the buffered ID type feature
${\mathbf{x}_{\xi}^{\mathrm{ID}}}$ according to the sample ID $\xi$; compute
gradients ${{F^{\mathrm{emb}}_{\xi}}}^{\prime}$ of the embedding parameters
and send the gradients to the embedding PS, so that the embedding PS can
finally compute the updates according the embedding parameter’s gradients by
its SGD optimizer and update the embedding parameters.

System implementation goals. To efficiently support the hybrid training
algorithm, we listed the following system design goals for Persia:

Fill the asynchronicity and synchronicity gap. One central functionality of
Persia system is to handle the heterogeneity inherited from the hybrid
training algorithm. This would request Persia to seamlessly connect the
forward- and backward- propagation during the training phase—the coordination
of embeddings and gradients transmitted in a large-scale cluster is a unique
challenge for Persia.

Utilize the heterogeneous clusters efficiently. To achieve the actual
performance gain, Persia needs to include different mechanisms to fully
utilize the computation resources (e.g., CPUs, GPUs) given diversified link
connections. This demands an efficient memory management and a group of
optimized communication mechanisms.

Provide effective fault tolerance. With the large number of machines that
training requires, effective fault tolerance is necessary. Furthermore, the
hybrid algorithm would request more complex mechanisms to manage the
heterogeneous computation and communication, which poses additional challenges
for fault tolerance.

###  4.2. System Optimizations and Design Decisions

We enumerate the implementation details to fulfill the design goals.

####  4.2.1. Fill the Async/Sync Gap

To fill the gap between synchronous and asynchronous updates, both embedding
and NN workers implement some buffering mechanisms.

NN worker buffer mechanism. Since we adopt a GPU-pull based schema for the
hybrid training procedure between the NN worker and the embedding worker, each
NN worker will locally maintain an input sample hash-map keyed on the sample
ID $\xi$ and valued on tuples of Non-ID type feature
${\mathbf{x}_{\xi}^{\mathrm{NID}}}$ and label $\mathbf{y}_{\xi}$. In the
forward propagation, once a NN worker receives the Non-ID type feature and
label, it will first insert the key-value pair
$\left(\mathrm{key}:\xi,\>\mathrm{value}:\left({\mathbf{x}_{\xi}^{\mathrm{NID}}},\mathbf{y}_{\xi}\right)\right)$
to the input sample hash-map; and then send the request of the embedding
vector to the embedding worker. Later, when the embedding vector
$\mathbf{w}^{\mathrm{emb}}_{\mathrm{\xi}}$ arrives in the NN worker from an
embedding worker, the NN worker will pop the key-value pair
$\left(\mathrm{key}:\xi,\>\mathrm{value}:\left({\mathbf{x}_{\xi}^{\mathrm{NID}}},\mathbf{y}_{\xi}\right)\right)$
from the input sample hash-map, consume
$\mathbf{w}^{\mathrm{emb}}_{\mathrm{\xi}}$,
${\mathbf{x}_{\xi}^{\mathrm{NID}}}$, $\mathbf{y}_{\xi}$ in the mini-batch for
the SGD computation within the GPU. In the backward propagation, once the
computation is done, the NN worker will use the sample ID $\xi$ to locate the
embedding worker, and then send the gradient of the embedding
${{F^{\mathrm{emb}}_{\xi}}}^{\prime}$ to this embedding worker.

Embedding worker buffering mechanism. To support the pull request from NN
worker, an embedding worker also needs to locally maintain a ID type feature
hash-map that is keyed on sample ID $\xi$ and valued on ID type feature
${\mathbf{x}_{\xi}^{\mathrm{ID}}}$. In the forward propagation, when the
embedding worker receives a new ID type feature, it will generate a unique ID
for this sample, store the key-value pair
$\left(\mathrm{key}:\xi,\>\mathrm{value}:{\mathbf{x}_{\xi}^{\mathrm{ID}}}\right)$
in the ID type feature hash-map. Then the embedding worker will get the
corresponding embedding parameter $\mathbf{w}^{\mathrm{emb}}_{\mathrm{\xi}}$
from embedding PS and return it to the NN worker. In the backward propagation,
once the embedding worker receives the gradient of the embedding vector
${{F^{\mathrm{emb}}_{\xi}}}^{\prime}$ from a NN worker, the embedding worker
will find the corresponding ID type feature
${\mathbf{x}_{\xi}^{\mathrm{ID}}}$. Note that this is implemented as a local
search in the ID type feature hash-map—the sample ID $\xi$ serves as the key
to retrieve ${\mathbf{x}_{\xi}^{\mathrm{ID}}}$. Later, the embedding worker
can put the gradient of the embedding parameters back to the embedding PS.

####  4.2.2. Persia Memory Management

Memory management is an important component for Persia to efficiently utilize
the hybrid infrastructure, especially for the embedding PS which is
responsible for maintaining trillions of embedding parameters.
Comprehensively, the embedding PS works like a standard PS, which is not
significantly different from a distributed key-value store. As we illustrated
in Figure 5, when retrieving the embedding parameters, an embedding worker
first runs an identical global hashing function to locate the embedding PS
node that stores the parameters; once the request arrives in the PS node, the
parameter can be acquired in an LRU cache as we explain below.

Figure 5. Memory management of Persia’s embedding PS node. The design is based
on a LRU cache implemented by hash-map and array-list.

LRU cache implementation. Persia leverages LRU cache444Standard LRU
implementation is based on hash-map and doubly linked list, e.g., (lru,
[n.d.]). to maintain the embedding parameters in RAM. We use array-list and
hash-map to implement the LRU. Instead of a doubly linked list where the
pointer stores a memory address, we adopt an array-list design where the
pointer stores the index of the pre- or post- entrance in the array;
similarly, the hash-map’s value also stores the corresponding embedding
parameter’s index in the array instead of the memory address. Besides the pre-
and post- indices, each item in the array also includes two fields: the
embedding vector and the optimizer states corresponding to this embedding
vector. There are two advantages by utilizing an array-based linked list: i)
this mechanism avoids frequent allocation and deallocation of fragmented
memory blocks—this cost is not negligible when each linked list may contain
billions of entrances; ii) since pointer (that stores the memory address) does
not exist in the data structure, serialization and deserialization become a
straightforward memory copy, which is far more convenient and efficient—this
is helpful for periodic saving and loading checkpoints for fault tolerance and
model sharing. Note that to fully utilize the CPU cores and RAM in the
embedding PS node, we utilize multiple threads in the LRU implementation. Each
thread manages a subset of the local hash-map and the corresponding array-
list; when there is a request of get or put, the corresponding thread will
lock its hash-map and array-list until the execution is completed.

####  4.2.3. Communication Optimization

To fully utilize the computation power in the heterogeneous cluster, Persia
implements a range of system optimizations to decrease the communication
overhead.

Optimized communication among NN workers. The optimization of AllReduce
communication paradigm among NN workers in Persia is the key for hiding
communication overhead within the backward computation of the neural network.
This functionality is implemented based on Bagua (Gan et al., 2021), an open-
source general-purposed distributed learning system optimized for data
parallelism, also released by Kwai. Currently, Persia utilizes Bagua’s
centralized synchronous full-precision communication primitive (equivalent to
AllReduce) by default, in an attempt to preserve the accuracy. Persia
leverages Bagua for the synchronization among NN workers because additional
system optimization enabled by Bagua can be directly adopted, including tensor
bucketing, memory flattening, hierarchical communications, etc. Notice that
there are other communication primitives provided by Bagua which could
potentially further improve the training throughput; however, it is unclear if
these communication primitives would hurt the statistical efficiency, we leave
this exploration as an interesting future work.

Optimized remote procedure call. The point-to-point communication: i) between
NN workers and embedding workers, and ii) between embedding workers and
embedding PS is implemented by remote procedure call (RPC). Unlike the
traditional usage of RPC, where the communication is mainly responsible for
transmitting small objects with complex serialization and deserialization
mechanism, Persia demands a RPC implementation that is efficient for
communicating tensors stored in a large continuous memory space. As so, Persia
abandons the protocol buffer based implementation (adopted by gRPC (grp,
[n.d.]) and bRPC (brp, [n.d.]) for other learning systems), which introduces
significant overhead for communicating tensors; instead, Persia adopts a
simple but efficient zero-copy serialization and deserialization mechanism
targeting for tensors which directly uses memory layout to serialize and
deserialize tensors. Further, as we mentioned in Section 4.2.2, since tensors
on host RAM are allocated in large pages, the TLB lookup time is also reduced
significantly, which accelerates the copying procedure from host RAM to the
RAM of the network adapter.

Workload balance of embedding PS. The embedding parameter storage is
implemented as a sharded PS to support the query and update of the parameter
in the embedding layer. The key challenge in embedding PS implementation is
about the workload balance of query and update about the embedding parameter.
Initially, we adopt a straightforward design by distributing the embedding
parameter according to the feature groups. A sub-group of CPU instances are
allocated to manage a partition of semantic independent embeddings. We find
that in practice, this would easily lead to congestion in the access of some
feature groups during training—the access of training data can irregularly
lean towards a particular embedding group during the online learning procedure
in industrial-scale applications. We solve this issue by adopting an
alternative partition of the embedding parameter: the embeddings inside a
feature group are first uniformly shuffled and then evenly distributed across
embedding PS nodes. We observe that this design effectively diminishes the
congestion of the embedding parameter access and keeps a balanced workload for
the embedding PS.

Communication compression. To reduce the network traffic, we adopt both
lossless and lossy compression mechanisms for the communication request
between embedding and NN workers. The network bandwidth connecting GPU
instances is limited—besides the AllReduce operation that leverages this
bandwidth, another noticeable utility is the communication of embedding
vectors in the forward propagation (\small{4}⃝ in Figure 4) and its gradient
in the backward propagation (\small{6}⃝ in Figure 4). The communication
between GPUs enabled by advanced connections like GPUDirect RDMA is $10\times$
faster than that between GPU and CPU nodes through PCI-e (gpu, [n.d.]), which
could leave the communication of embedding activation and its gradients as a
bottleneck. Note that although plenty of lossy compression schema has been
proposed for distributed learning, one should be cautious of applying them in
distributed recommender model training—for a commercial recommender system, as
mentioned before, even a drop of $0.1\%$ accuracy is not affordable (Zhao et
al., 2020b). To reduce such network traffic, we apply a lossless compression
mechanism for the index component and a discreet lossy compression mechanism
for the value component.

Lossless compression. For the index component of the embedding, instead of
representing a batch of samples as a list of vectors, where each vector
containing all IDs (represented by int64) of a sample, we represent a batch as
a hash-map, where the key is unique IDs in the whole batch, and the value
corresponding to each unique ID is the indices of the samples in the batch
containing this ID. Since the batch size is relatively small ($\leq 65535$),
the indices can be represented using uint16 instead of int64 without losing
any information.

Lossy compression. For the value component, We adopt a discreet fp32 to fp16
compression. Notice that a uniform mapping from fp32 to fp16 would harm the
statistic efficiency significantly, so we defines an nonuniform mapping
method: suppose $\lVert\cdot\rVert_{\infty}$ represents the $L_{\infty}$ norm
of a vector, $\kappa$ represents a relatively large constant scalar. In the
compression side, each fp32 vector block $\mathbf{v}$ is first scaled by
$\frac{\kappa}{\lVert\mathbf{v}\rVert_{\infty}}$ and then converted to a fp16
block vector. In the decompression side, the compressed block vector
$\tilde{\mathbf{v}}$ communicated as fp16 is first converted back to a fp32
vector and then divided by $\frac{\kappa}{\lVert\mathbf{v}\rVert_{\infty}}$.

####  4.2.4. Fault Tolerance

Failure can happen frequently considering the large number of nodes that
participate in the hybrid training. Towards this end, Persia implements a
group of fault tolerance mechanisms to handle different failures in the
cluster. Two observations are interesting:

  * •

The infrequent loss of parameter update of the embedding layer is usually
negligible for convergence, while the responsive time of the embedding
parameter query would be important for the end-to-end training time;

  * •

Any drop of the model synchronization by the NN worker that runs the dense
synchronous training algorithm is vital for convergence. Based on these
observations, Persia implements the following mechanisms for different
components to handle different system failures.

Since the data loader should be able to run on top of any other popular
distributed storage systems (e.g., HDFS), Persia relies on their own recovery
schema once an instance failure happens. Notice that Persia mainly considers
the online training setting, where no shuffling schema is required by Persia’s
data loader.

The embedding PS should be responsive during the hybrid training execution.
For this purpose, the embedding PS node will put in-memory LRU cache
(introduced in Section 4.2.2) in a shared memory space—by this fashion, once a
process-level failure happens, the process can automatically restart and
attach to the consistent shared-memory space without influencing any other
instances of the embedding PS. Additionally, embedding PS nodes will
periodically save the in-memory copy of the embedding parameter shard, with
the advance of our LRU implementation, check-pointing is very efficient (also
see Section 4.2.2).

The embedding worker has no fault recovery schema—once a failure happens, the
local buffer of the ID type feature hash-map will be simply abandoned without
any recovery attempts.

By contrast, the dense module cannot afford any drop of model synchronization.
As so, the NN worker would also periodically save the synchronized model as
the checkpoint with the same frequency as the embedding PS. Once a failure of
GPU instances happens, all the GPU instances will abandon their local copy of
the model, load the latest checkpoint, and continue the execution of the dense
synchronous training algorithm.

##  5\. Theoretical Analysis

In this section, we show the convergence rate of the proposed hybrid algorithm
adopted by Persia. In short, the proposed hybrid algorithm converges and
admits a similar convergence rate (or total complexity equivalently) to the
standard synchronous algorithm. That is to say that the hybrid algorithm takes
a significant benefit from asynchronicity in the system efficiency but
sacrifices very little for the convergence rate.

To show the detailed convergence rate, we first provide the essential updating
rule of the hybrid algorithm in Persia:

(2) $\displaystyle\mathbf{w}^{\mathrm{emb}}_{\mathrm{t+1}}=$
$\displaystyle\mathbf{w}^{\mathrm{emb}}_{\mathrm{t}}-\gamma{{F^{\mathrm{emb}}_{t}}}^{\prime}$
$\displaystyle\mathbf{w}^{\mathrm{nn}}_{\mathrm{t+1}}=$
$\displaystyle\mathbf{w}^{\mathrm{nn}}_{\mathrm{t}}-\gamma{{F^{\mathrm{nn}}_{t}}}^{\prime},$

where ${{F^{\mathrm{emb}}_{t}}}^{\prime}$ and
${{F^{\mathrm{nn}}_{t}}}^{\prime}$ are notations for gradients of
$\mathbf{w}^{\mathrm{emb}}_{\mathrm{t}}$ and
$\mathbf{w}^{\mathrm{nn}}_{\mathrm{t}}$, short for respectively:

$\displaystyle{{F^{\mathrm{emb}}_{t}}}^{\prime}:=$
$\displaystyle\nabla_{\mathbf{w}^{\mathrm{emb}}}F\left(\mathbf{w}^{\mathrm{emb}}_{\mathrm{\mathrm{D}(t)}},\mathbf{w}^{\mathrm{nn}}_{\mathrm{t}};\xi_{t}\right),$
$\displaystyle{{F^{\mathrm{nn}}_{t}}}^{\prime}:=$
$\displaystyle\nabla_{\mathbf{w}^{\mathrm{nn}}}F\left(\mathbf{w}^{\mathrm{emb}}_{\mathrm{\mathrm{D}(t)}},\mathbf{w}^{\mathrm{nn}}_{\mathrm{t}};\xi_{t}\right),$

where $D(t)\leqslant t$ denotes some iterate earlier than the current iterate
$t$. This is because of the asynchronous update for the embedding layer. Since
the dense neural network adopts the synchronous update, it always uses the up
to date value of $\mathbf{w}^{\mathrm{nn}}_{\mathrm{t}}$, while the used value
of the embedding layer may be from some early iterate. This is the cost of
using asynchronous updating, but we will see that the cost is very minor
comparing to our gain in system efficiency.

To ensure the convergence rate, we still need to make some commonly used
assumptions as follows:

###### Assumption 1.

We make the following commonly used assumptions for analyzing the stochastic
algorithm:

  * •

Existence of global minimum. Assume

$f^{*}:=\min_{\mathbf{w}}f\left(\mathbf{w}\right)$

exists.

  * •

Lipschitzian gradient. The stochastic gradient function
$F^{\prime}\left(\cdot;\xi\right)$ is differentiable, and $L$-Lipschitzian for
all $\xi$:

$\displaystyle\left\|F^{\prime}\left(\mathbf{w};\xi\right)-F^{\prime}\left(\bar{\mathbf{w}};\xi\right)\right\|\leqslant
L\left\|\mathbf{w}-\bar{\mathbf{w}}\right\|,$ (3)
$\displaystyle\quad\forall\xi\in\Xi,\forall\mathbf{w},\bar{\mathbf{w}}\in\mathbb{R}^{N^{\mathrm{emb}}+N^{\mathrm{nn}}}.$

  * •

Bounded variance. The variance of the stochastic gradient is bounded: there
exists a constant $\sigma\geqslant 0$ such that

(4)
$\displaystyle\mathbb{E}\left[\left\|F^{\prime}\left(\mathbf{w};\xi\right)-f^{\prime}\left(\mathbf{w}\right)\right\|^{2}\right]\leqslant\sigma^{2}\quad\forall\mathbf{w}.$

  * •

Bounded staleness. All sparse model update delays are bounded: there exists a
constant $\tau\geqslant 0$ such that

(5) $\displaystyle t-D(t)\leqslant$ $\displaystyle\tau,\quad\forall t.$

In practice $\tau$ is the number of samples whose embeddings are retrieved but
their gradients are not yet updated into the model parameters. In Persia this
value is less than 5 for most cases.

###### Remark 1.

The assumptions of existence, bounded variance, and bounded staleness are
commonly used ones. The bounded staleness assumption is due to the
asynchronous update for $\mathbf{w}^{\mathrm{emb}}$. In practice,
$t-D(t)\leqslant\tau$ means that any element in $\mathbf{w}^{\mathrm{emb}}$ is
updated up to $\tau$ times between reading it and writing it within training
the same sample. In Persia this value of $\tau$ is less than 5 for most
scenarios. The frequency assumption reflects the intrinsic property of the
recommendation system. For each sample $\xi$, if every ID type feature $x$ has
the same probability of being in ${\mathbf{x}_{\xi}^{\mathrm{ID}}}$, then
$\alpha=\frac{1}{N^{\mathrm{emb}}}$. On the other hand, if an ID type feature
$x$ is contained by every sample $\xi$’s ID type features
${\mathbf{x}_{\xi}^{\mathrm{ID}}}$, then $\alpha=1$.

Then we obtain the following convergence result:

###### Theorem 1.

Denote $\alpha$ to be the constant such that for each ID the probability of a
sample containing it is smaller than $\alpha$. Under Assumption 1, with
learning rate $\gamma$ in (2) as

$\displaystyle\gamma=\frac{1}{L+\sqrt{TL}\sigma+4\tau L\alpha}$

Persia admits the following convergence rate:

(6)
$\displaystyle\frac{\sum_{t=0}^{T-1}\mathbb{E}\left[\left\|f^{\prime}(\mathbf{w}_{t})\right\|^{2}\right]}{T}\lesssim$
$\displaystyle\underbrace{\frac{\sigma}{\sqrt{T}}+\frac{1}{T}+}_{\text{rate of
vanilla SGD}}\frac{\alpha\tau}{T}$

where $\lesssim$ means “smaller than or equal to up to a constant factor”, and
$L,G$ are all treated as constants for simplicity. For example $a_{t}\lesssim
b_{t}$ means that there exists a constant $C>0$ such that $a_{t}\leq Cb_{t}$
for all $t$.

The complete proof is provided in the preprint version of this paper (Lian et
al., 2021). Note that the first two terms in (6) are exactly the convergence
rate of vanilla SGD (Liu and Zhang, 2020). The third (additional) term is
caused by the staleness of the asynchronous update in the embedding layer of
the model. The staleness upper bound $\tau$ is typically proportional to the
number of workers. The ID frequency upper bound $\alpha$ is a value smaller
than $1$. Notice that $\alpha=1$ leads a convergence rate exactly matching the
asynchronous SGD (Liu and Zhang, 2020). It indicates that Persia (the hybrid
algorithm) guarantees to be no worse than the asynchronous algorithm in
convergence rate. When $\alpha\ll 1$, which is mostly always true in real
recommendation systems, the hybrid algorithm—Persia—admits a very similar
convergence rate to the synchronous algorithm, since the third term is
dominated by the second term. Therefore, this theorem suggests that the hybrid
algorithm follows the same convergence efficiency in terms of iterations.

Figure 6. End-to-end training performance of four benchmarks: Taobao-Ad,
Avazu-Ad, Criteo-Ad, Kwai-Video (from left to right).

##  6\. Experiments

In this section, we introduce experiments to evaluate the design and
implementation of Persia, focusing on the following questions:

  * •

Can Persia provide significant boost of end-to-end training time comparing
with the state-of-the-art systems?

  * •

Is the hybrid training algorithm statistically efficient in terms of
convergence w.r.t training iterations?

  * •

Can Persia provide high training throughput and scale out well especially for
trillion-parameter scale models?

Benchmark. We evaluate Persia over three open-source benchmarks and one real-
world production microvideo recommendation workflow at Kwai:

  * •

Taobao-Ad (open source (tao, [n.d.])): predict the advertisement CTR from
Taobao’s website for 8 days with 26 million records.

  * •

Avazu-Ad (open source (ava, [n.d.])): predict the advertisement CTR of Avazu’s
log for 11 days with 32 million records.

  * •

Criteo-Ad (open source (cri, [n.d.])): predict the advertisement CTR of
Criteo’s traffic for 24 days with 44 million records. Note that we also extend
this dataset (noted as Criteo-Syn) by synthesizing different number of random
ID features for capacity and scalablity evaluation (see Section 6.3).

  * •

Kwai-Video (confidential production dataset): predict the explicit behavior of
Kwai’s active users about the microvideo recommendation in $7$ days with 3
billion records.

For the three open source advertisement CTR benchmarks, we include $80\%$ of
the records as training set and the rest $20\%$ of the records as test set, we
consider a fully connected feed forward neural network (FFNN) as the deep
learning model with five hidden layer dimensions of $4096$, $2048$, $1024$,
$512$ and $256$. For the Kwai production microvideo recommendation task,
$85\%$ of the data are included in the training set while the rest $15\%$ are
considered as the test set, we also use FFNN as the model to predict multiple
user behaviors. We report test AUC to evaluate convergence. The model scale of
the each benchmark is listed in Table 1

| Sparse # parameter | Dense # parameter
---|---|---
Taobao-Ad | 29 Million | 12 Million
Avazu-Ad | 134 Million | 12 Million
Criteo-Ad | 540 Million | 12 Million
Kwai-Video | 2 Trillion | 34 Million
Criteo-Syn1 | 6.25 Trillion | 12 Million
Criteo-Syn2 | 12.5 Trillion | 12 Million
Criteo-Syn3 | 25 Trillion | 12 Million
Criteo-Syn4 | 50 Trillion | 12 Million
Criteo-Syn5 | 100 Trillion | 12 Million
Table 1. Model scales for the benchmarks.

Baseline systems. We consider two state-of-the-art baselines: XDL (Jiang et
al., 2019) and PaddlePaddle (pad, [n.d.])—XDL is a specialized distributed
recommendation framework developed by Alibaba; PaddlePaddle is a general
purpose deep learning framework from Baidu with a special Heter mode that
implements the design of AIBox (Zhao et al., 2019b), Baidu’s recommendation
training system, according to private communications we had with members of
the PaddlePaddle development community.

Cluster setup. We conduct experiments on two clusters. Most of the training is
conducted over heterogeneous clusters inside Kwai’s production data center—for
Persia, we include up to $64$ Nvidia V100 GPUs, and $100$ CPU instances (each
with $52$ cores and $480$GB RAM). The instances are connected by a network
with the bandwidth of 100 Gbps. The baseline systems (XDL and PaddlePaddle)
are equipped with the same amount of computation resources for each individual
setting.

We further conduct larger-scale scalability experiments over Google cloud
platform, for both capacity and public reproducibility reasons555We appreciate
Google’s great support for coordinating the computation resources for this
experiment. with a heterogeneous cluster including:

  * •

8 a2-highgpu-8g instances (each with 8 Nvidia A100 GPUs) as NN workers;

  * •

100 c2-standard-30 instances (each with 30vCPUs, 120GB RAM) as embedding
workers;

  * •

30 m2-ultramem-416 instances (each with 416vCPUs, 12TB RAM) as embedding PS.

###  6.1. End-to-end Evaluation

We first compare the end-to-end training time that each system needs to
achieve a given AUC, using the three open source benchmarks over a
heterogeneous cluster with $8$ GPU workers. We report the results of both
synchronous and asynchronous modes of XDL, we conduct a careful trial of
different modes in PaddlePaddle and report the result under the optimal
setting. Figure 6 illustrates significant performance improvements from
Persia: e.g., for the Taobao-Ad benchmark, Persia is $7.12\times$ and
$8.4\times$ faster than that of the synchronous and asynchronous modes of XDL,
and $1.71\times$ faster than PaddlePaddle–same level of speedup also appears
in the Avazu-Ad and Criteo-Ad benchmark.

On our production Kwai-Video dataset, XDL fails to run the task to convergence
in $100$ hours—we compare the training sample throughput and find that
Persia’s throughput is $19.31\times$ of XDL). For PaddlePaddle, we fail to run
the training task since PaddlePaddle does not support some deep learning
operators (e.g., batch normalization) required in our model.

###  6.2. Evaluation of Hybrid Algorithm

Figure 7. Convergence of four benchmarks: Taobao-Ad, Avazu-Ad, Criteo-Ad, Kwai-Video. |  Persia-Hybrid | Sync | Async | PaddlePaddle
---|---|---|---|---
Taobao-Ad | 65.86% | 65.93% (XDL) | 65.66% (XDL) | 59.14%
Avazu-Ad | 77.93% | 78.02% (XDL) | 77.10% (XDL) | 73.19%
Criteo-Ad | 80.47% | 80.51% (XDL) | 80.01% (XDL) | 77.08%
Kwai-Video | 72.63% | 72.63% | 72.16% | -
Table 2. Final test AUC of four benchmarks.

The hybrid algorithm is one of the main reasons behind Persia’s scalability.
Here, we evaluate its convergence behavior and leave it scalability and
performance to the next section. Figure 7 illustrates the convergence
behaviors of different systems. We see that the hybrid algorithm shows almost
identical convergence when comparing with the fully synchronous mode,
converging to comparable AUC as illustrated in Table 2 (but as we will see in
the next section, the hybrid algorithm is much faster compared with the fully
synchronous algorithm). We see that test AUC gap between the hybrid mode and
synchronous mode is always less than $0.1\%$ in the three open-source
benchmarks, and less than $0.001\%$ in the production Kwai-video benchmark; by
contrast, the gap between the asynchronous mode and the synchronous mode is
much higher (from $0.5\%$ to $1.0\%$); further, as we allow more aggressive
asynchronicity in PaddlePaddle, the gap is more significant. We want to
emphasize that even $0.1\%$ decrease of accuracy for the production
recommender model might lead to a loss of revenue at the scale of a million
dollar—this is NOT acceptable.

Figure 8. Scalability and performance of four benchmarks: Taobao-Ad, Avazu-Ad,
Criteo-Ad, Kwai-Video (from left to right).  Figure 9. Capacity test of Persia
for Criteo-Syn benchmark the over Google cloud platform.

###  6.3. Scalability and Capacity

We first evaluate the scalability of Persia over clusters with different
scales of GPUs in terms of training sample throughput, as illustrated in
Figure 8. We see that Persia, with a hybrid algorithm, achieves much higher
throughput compared to all other systems. On three open source benchmarks,
Persia reaches nearly linear speedup with significantly higher throughput
comparing with XDL and PaddlePaddle. For the Kwai-video benchmark, Persia
achieves $3.8\times$ higher throughput compared with the fully synchronous
algorithm. Note that the fully asynchronous algorithm can achieve faster
throughput than the hybrid algorithm; however, as illustrated in the previous
section, it incurs a lower AUC compared with the hybrid algorithm.

We further investigate the capacity of Persia for training different scales of
trillion-parameter models in Google cloud platform. Figure 9-left illustrates
that the throughput of Persia when varying the number of entrances in the
embedding table and fixing the embedding output dimension to $128$—this is
corresponding to the model scale of Criteo-Syn in Table 1. We see that Persia
shows stable training throughput when increasing the model size even up to 100
trillion parameters. This can be viewed as concrete evidences that Persia has
the capacity to scale out to the largest recommender model that is never
reported before. Figure 9-right shows that for the 100 trillion-parameter
model, Persia also achieves $2.6\times$ higher throughput than the fully
synchronous mode; on the other hand, asynchronous mode introduces further
speedup ($1.2\times$ faster than that of the hybrid algorithm)—this generally
would introduce statistical inefficiency for convergence, which is not
revealed by the capacity evaluation.

##  7\. Related Work

We briefly discuss the relevant recommender systems and system relaxations for
distributed large scale learning. Detailed discussion can be found in more
comprehensive surveys (e.g., (Zhang et al., 2019c) for recommender system, and
(Liu and Zhang, 2020) for distributed deep learning).

###  7.1. Recommender Systems

Recommender systems are critical tools to enhance user experience and promote
sales and services (Zhang et al., 2019c), where deep learning approaches have
shown great advance recently and been deployed ubiquitously by tech companies
such as Alibaba (Zhou et al., 2019), Amazon (Raman et al., 2019), Baidu (Zhao
et al., 2020b), Facebook (Naumov et al., 2020), Google (Covington et al.,
2016; Zhao et al., 2019a), Kuaishou (Xie et al., 2020), Pinterest (Liu et al.,
2017), Netflix (Gomez-Uribe and Hunt, 2015), etc. Generally, deep learning
based approaches take the sparse ID type features to generate dense vectors
through an embedding layer, and then feed the embedding activations into the
proceeding neural components to expose the correlation and interaction.
Diversified features are included to power recommender systems (Zhang et al.,
2017a), such as sequential information (e.g., session based user id (Tan et
al., 2016; Twardowski, 2016; Tuan and Phuong, 2017)), text (Zuo et al., 2016;
Zhang et al., 2017c; Zhao et al., 2018), image (Wang et al., 2017b; Wen et
al., 2018), audio (Van Den Oord et al., 2013; Wang and Wang, 2014), video
(Chen et al., 2017; Lee et al., 2018), social network (Hsieh et al., 2016;
Deng et al., 2016), etc. Additionally, various neural components have been
explored for recommender systems, including multi-layer perceptron (MLP) (Xu
et al., 2016; Yang et al., 2017), convolutional neural network (CNN) (Seo et
al., 2017; Tuan and Phuong, 2017), recurrent neural network (RNN) (Wu et al.,
2016b, a), Autoencoder (Sedhain et al., 2015; Zhuang et al., 2017), and deep
reinforcement learning (Zhao et al., 2018; Zheng et al., 2018). Furthermore,
multiple relevant goals for recommendation can be learned simultaneously by
multi-task learning (Covington et al., 2016; Zhou et al., 2018b, 2019; Zhang
et al., 2020; Naumov et al., 2020; Zhao et al., 2020a).

Due to the massive scale of both training data and recommender models,
distributed learning techniques are applied to train industrial-scale
recommender systems (Zhou et al., 2019; Raman et al., 2019; Zhao et al.,
2020b; Zhao et al., 2019a; Naumov et al., 2020). The heterogeneity in
computation naturally leads to the idea of utilizing heterogeneous distributed
computation resources to handle such training procedure (Jiang et al., 2019;
Raman et al., 2019; Zhao et al., 2019b; Pan et al., 2020; Zhao et al., 2020b;
Naumov et al., 2020; Krishna and Krishna, 2020; Wilkening et al., 2021; Miao
et al., 2022). For example, XDL (Jiang et al., 2019) adopts the advanced model
server to replace classic parameter server to reduce the traffic between CPU
nodes and GPU nodes; Baidu proposes a hierarchical PS architecture (Zhao et
al., 2020b) to implement a sophisticated caching schema to reduce the
communication overhead; HET (Miao et al., 2022) adopts an advanced cache
mechanism at the GPU worker side to leverage the skewness of embeddings by
caching frequently updated entrances in the worker’s limited local memory.
However, as far as we know, there is a lack of a carefully designed
distributed training solution to support the heterogeneity inherited from the
training data (sparse vs. dense) and gradient decent based optimization
(memory bounded vs. computation bounded) over mixed hardwares and
infrastructures (CPU instances vs. GPU instances).

###  7.2. Distributed Deep Learning

Distributed deep learning can be categorized as data parallelism and model
parallelism. In data parallel distributed deep learning, two main categories
of systems are designed—parameter server (Dean et al., 2012; Li et al., 2014;
Huang et al., 2018; Jiang et al., 2017; Zhang et al., 2019a) and AllReduce
(Sergeev and Del Balso, 2018; Zhang et al., 2019b; Jiang et al., 2018; Chen
and Guestrin, 2016). In a PS architecture, models are stored in a single node
or distributively in multiple nodes; during the training phase, workers
periodically fetch the model from PS, conduct forward/backward propagation,
and push the gradients to the PS, while the PS aggregates the gradients and
updates the parameters. With an AllReduce paradigm, all workers collaborate
with their neighbors for model/gradient exchanges. Different strategies are
proposed to speed up the expensive parameter/gradient exchange phase. To
reduce communication volumes, lossy communication compression methods are
introduced, such as quantization (Alistarh et al., 2016; Zhang et al., 2017b;
Bernstein et al., 2018; Wen et al., 2017), sparsification (Wangni et al.,
2018; Alistarh et al., 2018; Wang et al., 2018b; Wang et al., 2017a),
sketching (Ivkin et al., 2019), and error compensation (Tang et al., 2019b)).
In an attempt to get rid of the latency bottleneck, decentralized
communication approaches are proposed (Koloskova et al., 2019; Li et al.,
2018a; Lian et al., 2017; Lian et al., 2018; Tang et al., 2018). Additionally,
local SGD is discussed to optimize for the number of communication rounds
during training (Wang and Joshi, 2019; Lin et al., 2019; Stich, 2018;
Haddadpour et al., 2019). To remove the synchronization barrier, asynchronous
update methods are proposed (Lian et al., 2015; Peng et al., 2017; Zheng et
al., 2017; Zhou et al., 2018a; Simsekli et al., 2018; Nguyen et al., 2018).
There are also approaches that combines multiple strategies listed above (Lian
et al., 2018; Basu et al., 2019; Koloskova et al., 2019; Tang et al., 2019a;
Beznosikov et al., 2020). On the other hand, researches about model
parallelism attempt to study how to allocate model parameters and training
computation across compute units in a cluster to maximize training throughput
and minimize communication overheads. Optimizations are proposed for both
operation partitioning approach (Jia et al., 2019; Wang et al., 2019; Shazeer
et al., 2018; Shoeybi et al., 2019) and pipeline parallel approach (Huang et
al., 2019; Harlap et al., 2018; Tarnawski et al., 2020). Recently, there are
also approaches that combine both data and model parallelism (Raman et al.,
2019; Park et al., 2020; Li et al., 2021).

General purpose deep learning systems have been one of the main driving forces
behind the rapid advancement of machine learning techniques with the
increasing scalability and performance for distributed deep learning. Popular
options include TensorFlow (Abadi et al., 2016), PyTorch (pyt, [n.d.]), MXNet
(Chen et al., 2015), PaddlePaddle (pad, [n.d.]), MindSpore (min, [n.d.]), etc.
Extensions and modifications have been made based on these general purpose
learning systems for efficient distributed learning (e.g., Horovod (Sergeev
and Del Balso, 2018), BytePS (Jiang et al., 2020), Bagua (Gan et al., 2021),
Megatron (Shoeybi et al., 2019), ZeRO (Rajbhandari et al., 2020), SageMaker
(Karakus et al., 2021), etc.). However, even including these extensions, the
current general purpose deep learning systems do not consider the challenges
about handling the heterogeneity over a hybrid infrastructure. Thus, it is
difficult to directly use these general purpose learning systems for the
training tasks of industrial-scale recommender systems over a heterogeneous
cluster.

##  8\. Conclusion

We introduce Persia, a distributed system to support efficient and scalable
training of recommender models at the scale of 100 trillion parameters. We
archive the statistical and hardware efficiency by a careful co-design of both
the distributed training algorithm and the distributed training system. Our
proposed hybrid distributed training algorithm introduces elaborate system
relaxations for efficient utilization of heterogeneous clusters, while
converging as fast as vanilla SGD. We implement a wide range of system
optimizations to overcome the unique challenges raised by the hybrid
algorithmic design. We evaluate Persia using both publicly available benchmark
tasks and production tasks at Kwai. We show that Persia leads to up to
$7.12\times$ speedup compared to alternative state-of-the-art approaches;
additionally, Persia can also scale out to 100-trillion-parameter model on
Google cloud platform.

Persia has been released as an open source project on github with concrete
instructions about setup over Google cloud platform—we hope everyone from both
academia and industry would find it easy to train 100-trillion-parameter scale
deep learning recommender models.

## References

  * (1)
  * ava ([n.d.]) [n.d.].  Avazu Advertisement CTR.  https://www.kaggle.com/c/avazu-ctr-prediction.
  * brp ([n.d.]) [n.d.].  bRPC Industrial-grade RPC framework used throughout Baidu.  http://incubator.apache.org/clutch/brpc.html.
  * cri ([n.d.]) [n.d.].  Criteo Advertisement CTR.  https://ailab.criteo.com/download-criteo-1tb-click-logs-dataset/.
  * grp ([n.d.]) [n.d.].  gRPC A high performance, open source universal RPC framework.  https://grpc.io/.
  * lru ([n.d.]) [n.d.].  How to implement LRU cache using HashMap and Doubly Linked List.  https://krishankantsinghal.medium.com/my-first-blog-on-medium-583159139237.
  * min ([n.d.]) [n.d.].  MindSpore AI computing Framework Bridge the gap between AI research and application.  https://www.paddlepaddle.org.cn/.
  * gpu ([n.d.]) [n.d.].  NVIDIA GPUDirect.  https://developer.nvidia.com/gpudirect.
  * pyt ([n.d.]) [n.d.].  An open source machine learning framework that accelerates the path from research prototyping to production deployment.  https://pytorch.org/.
  * pad ([n.d.]) [n.d.].  Parallel Distributed Deep Learning: Machine Learning Framework from Industrial Practice.  https://www.paddlepaddle.org.cn/.
  * tao ([n.d.]) [n.d.].  Taobao Advertisement CTR.  https://www.kaggle.com/pavansanagapati/ad-displayclick-data-on-taobaocom.
  * Abadi et al. (2016) Martín Abadi, Paul Barham, Jianmin Chen, Zhifeng Chen, Andy Davis, Jeffrey Dean, Matthieu Devin, Sanjay Ghemawat, Geoffrey Irving, Michael Isard, et al. 2016\.  Tensorflow: A system for large-scale machine learning. In _12th $\\{$USENIX$\\}$ symposium on operating systems design and implementation ($\\{$OSDI$\\}$ 16)_. 265–283.
  * Acun et al. (2020) Bilge Acun, Matthew Murphy, Xiaodong Wang, Jade Nie, Carole-Jean Wu, and Kim Hazelwood. 2020\.  Understanding Training Efficiency of Deep Learning Recommendation Models at Scale.  _arXiv preprint arXiv:2011.05497_ (2020).
  * Alistarh et al. (2016) Dan Alistarh, Demjan Grubic, Jerry Li, Ryota Tomioka, and Milan Vojnovic. 2016.  QSGD: Communication-efficient SGD via gradient quantization and encoding.  _arXiv preprint arXiv:1610.02132_ (2016).
  * Alistarh et al. (2018) Dan Alistarh, Torsten Hoefler, Mikael Johansson, Sarit Khirirat, Nikola Konstantinov, and Cédric Renggli. 2018.  The convergence of sparsified gradient methods. In _Proceedings of the 32nd International Conference on Neural Information Processing Systems_. 5977–5987.
  * Basu et al. (2019) Debraj Basu, Deepesh Data, Can Karakus, and Suhas Diggavi. 2019\.  Qsparse-local-SGD: Distributed SGD with quantization, sparsification, and local computations.  _arXiv preprint arXiv:1906.02367_ (2019).
  * Bernstein et al. (2018) Jeremy Bernstein, Yu-Xiang Wang, Kamyar Azizzadenesheli, and Animashree Anandkumar. 2018.  signSGD: Compressed optimisation for non-convex problems. In _International Conference on Machine Learning_. PMLR, 560–569.
  * Beznosikov et al. (2020) Aleksandr Beznosikov, Samuel Horváth, Peter Richtárik, and Mher Safaryan. 2020.  On biased compression for distributed learning.  _arXiv preprint arXiv:2002.12410_ (2020).
  * Chang et al. (2021) Wei-Cheng Chang, Daniel Jiang, Hsiang-Fu Yu, Choon Hui Teo, Jiong Zhang, Kai Zhong, Kedarnath Kolluri, Qie Hu, Nikhil Shandilya, Vyacheslav Ievgrafov, et al. 2021\.  Extreme multi-label learning for semantic matching in product search. In _Proceedings of the 27th ACM SIGKDD Conference on Knowledge Discovery & Data Mining_. 2643–2651.
  * Chen et al. (2016) Jianmin Chen, Xinghao Pan, Rajat Monga, Samy Bengio, and Rafal Jozefowicz. 2016.  Revisiting distributed synchronous SGD.  _arXiv preprint arXiv:1604.00981_ (2016).
  * Chen et al. (2017) Jingyuan Chen, Hanwang Zhang, Xiangnan He, Liqiang Nie, Wei Liu, and Tat-Seng Chua. 2017\.  Attentive collaborative filtering: Multimedia recommendation with item-and component-level attention. In _Proceedings of the 40th International ACM SIGIR conference on Research and Development in Information Retrieval_. 335–344.
  * Chen and Guestrin (2016) Tianqi Chen and Carlos Guestrin. 2016.  Xgboost: A scalable tree boosting system. In _Proceedings of the 22nd acm sigkdd international conference on knowledge discovery and data mining_. 785–794.
  * Chen et al. (2015) Tianqi Chen, Mu Li, Yutian Li, Min Lin, Naiyan Wang, Minjie Wang, Tianjun Xiao, Bing Xu, Chiyuan Zhang, and Zheng Zhang. 2015\.  Mxnet: A flexible and efficient machine learning library for heterogeneous distributed systems.  _arXiv preprint arXiv:1512.01274_ (2015).
  * Cheng et al. (2016) Heng-Tze Cheng, Levent Koc, Jeremiah Harmsen, Tal Shaked, Tushar Chandra, Hrishi Aradhye, Glen Anderson, Greg Corrado, Wei Chai, Mustafa Ispir, et al. 2016\.  Wide & deep learning for recommender systems. In _Proceedings of the 1st workshop on deep learning for recommender systems_. 7–10.
  * Covington et al. (2016) Paul Covington, Jay Adams, and Emre Sargin. 2016\.  Deep neural networks for youtube recommendations. In _Proceedings of the 10th ACM conference on recommender systems_. 191–198.
  * Davidson et al. (2010) James Davidson, Benjamin Liebald, Junning Liu, Palash Nandy, Taylor Van Vleet, Ullas Gargi, Sujoy Gupta, Yu He, Mike Lambert, Blake Livingston, et al. 2010\.  The YouTube video recommendation system. In _Proceedings of the fourth ACM conference on Recommender systems_. 293–296.
  * Dean et al. (2012) Jeffrey Dean, Greg S Corrado, Rajat Monga, Kai Chen, Matthieu Devin, Quoc V Le, Mark Z Mao, Marc’Aurelio Ranzato, Andrew Senior, Paul Tucker, et al. 2012\.  Large scale distributed deep networks. In _Proceedings of the 25th International Conference on Neural Information Processing Systems-Volume 1_. 1223–1231.
  * Deng et al. (2016) Shuiguang Deng, Longtao Huang, Guandong Xu, Xindong Wu, and Zhaohui Wu. 2016.  On deep learning for trust-aware recommendations in social networks.  _IEEE transactions on neural networks and learning systems_ 28, 5 (2016), 1164–1177.
  * Eksombatchai et al. (2018) Chantat Eksombatchai, Pranav Jindal, Jerry Zitao Liu, Yuchen Liu, Rahul Sharma, Charles Sugnet, Mark Ulrich, and Jure Leskovec. 2018.  Pixie: A system for recommending 3+ billion items to 200+ million users in real-time. In _Proceedings of the 2018 world wide web conference_. 1775–1784.
  * Gan et al. (2021) Shaoduo Gan, Xiangru Lian, Rui Wang, Jianbin Chang, Chengjun Liu, Hongmei Shi, Shengzhuo Zhang, Xianghong Li, Tengxu Sun, Jiawei Jiang, et al. 2021\.  BAGUA: Scaling up Distributed Learning with System Relaxations.  _arXiv preprint arXiv:2107.01499_ (2021).
  * Gomez-Uribe and Hunt (2015) Carlos A Gomez-Uribe and Neil Hunt. 2015.  The netflix recommender system: Algorithms, business value, and innovation.  _ACM Transactions on Management Information Systems (TMIS)_ 6, 4 (2015), 1–19.
  * Haddadpour et al. (2019) Farzin Haddadpour, Mohammad Mahdi Kamani, Mehrdad Mahdavi, and Viveck R Cadambe. 2019.  Local sgd with periodic averaging: Tighter analysis and adaptive synchronization.  _arXiv preprint arXiv:1910.13598_ (2019).
  * Harlap et al. (2018) Aaron Harlap, Deepak Narayanan, Amar Phanishayee, Vivek Seshadri, Nikhil Devanur, Greg Ganger, and Phil Gibbons. 2018.  Pipedream: Fast and efficient pipeline parallel dnn training.  _arXiv preprint arXiv:1806.03377_ (2018).
  * Hsieh et al. (2016) Cheng-Kang Hsieh, Longqi Yang, Honghao Wei, Mor Naaman, and Deborah Estrin. 2016.  Immersive recommendation: News and event recommendations using personal digital traces. In _Proceedings of the 25th International Conference on World Wide Web_. 51–62.
  * Huang et al. (2019) Yanping Huang, Youlong Cheng, Ankur Bapna, Orhan Firat, Dehao Chen, Mia Xu Chen, HyoukJoong Lee, Jiquan Ngiam, Quoc V Le, Yonghui Wu, et al. 2019\.  GPipe: Efficient Training of Giant Neural Networks using Pipeline Parallelism.. In _NeurIPS_.
  * Huang et al. (2018) Yuzhen Huang, Tatiana Jin, Yidi Wu, Zhenkun Cai, Xiao Yan, Fan Yang, Jinfeng Li, Yuying Guo, and James Cheng. 2018.  Flexps: Flexible parallelism control in parameter server architecture.  _Proceedings of the VLDB Endowment_ 11, 5 (2018), 566–579.
  * Ivkin et al. (2019) Nikita Ivkin, Daniel Rothchild, Enayat Ullah, Ion Stoica, Raman Arora, et al. 2019\.  Communication-efficient distributed sgd with sketching. In _Advances in Neural Information Processing Systems_. 13144–13154.
  * Jia et al. (2019) Zhihao Jia, Matei Zaharia, and Alex Aiken. 2019\.  Beyond Data and Model Parallelism for Deep Neural Networks.  _SysML 2019_ (2019).
  * Jiang et al. (2019) Biye Jiang, Chao Deng, Huimin Yi, Zelin Hu, Guorui Zhou, Yang Zheng, Sui Huang, Xinyang Guo, Dongyue Wang, Yue Song, et al. 2019\.  XDL: an industrial deep learning framework for high-dimensional sparse data. In _Proceedings of the 1st International Workshop on Deep Learning Practice for High-Dimensional Sparse Data_. 1–9.
  * Jiang et al. (2018) Jiawei Jiang, Bin Cui, Ce Zhang, and Fangcheng Fu. 2018\.  Dimboost: Boosting gradient boosting decision tree to higher dimensions. In _Proceedings of the 2018 International Conference on Management of Data_. 1363–1376.
  * Jiang et al. (2017) Jiawei Jiang, Bin Cui, Ce Zhang, and Lele Yu. 2017\.  Heterogeneity-aware distributed parameter servers. In _Proceedings of the 2017 ACM SIGMOD International Conference on Management of Data_. 463–478.
  * Jiang et al. (2020) Yimin Jiang, Yibo Zhu, Chang Lan, Bairen Yi, Yong Cui, and Chuanxiong Guo. 2020\.  A Unified Architecture for Accelerating Distributed $\\{$DNN$\\}$ Training in Heterogeneous GPU/CPU Clusters. In _14th $\\{$USENIX$\\}$ Symposium on Operating Systems Design and Implementation ($\\{$OSDI$\\}$ 20)_. 463–479.
  * Karakus et al. (2021) Can Karakus, Rahul Huilgol, Fei Wu, Anirudh Subramanian, Cade Daniel, Derya Cavdar, Teng Xu, Haohan Chen, Arash Rahnama, and Luis Quintela. 2021\.  Amazon SageMaker Model Parallelism: A General and Flexible Framework for Large Model Training.  _arXiv preprint arXiv:2111.05972_ (2021).
  * Kingma and Ba (2014) Diederik P Kingma and Jimmy Ba. 2014.  Adam: A method for stochastic optimization.  _arXiv preprint arXiv:1412.6980_ (2014).
  * Koloskova et al. (2019) Anastasia Koloskova, Sebastian Stich, and Martin Jaggi. 2019\.  Decentralized stochastic optimization and gossip algorithms with compressed communication. In _International Conference on Machine Learning_. PMLR, 3478–3487.
  * Krishna and Krishna (2020) Suresh Krishna and Ravi Krishna. 2020.  Accelerating Recommender Systems via Hardware” scale-in”.  _arXiv preprint arXiv:2009.05230_ (2020).
  * Lee et al. (2018) Joonseok Lee, Sami Abu-El-Haija, Balakrishnan Varadarajan, and Apostol Natsev. 2018.  Collaborative deep metric learning for video understanding. In _Proceedings of the 24th ACM SIGKDD International Conference on Knowledge Discovery & Data Mining_. 481–490.
  * Li et al. (2014) Mu Li, David G Andersen, Jun Woo Park, Alexander J Smola, Amr Ahmed, Vanja Josifovski, James Long, Eugene J Shekita, and Bor-Yiing Su. 2014.  Scaling distributed machine learning with the parameter server. In _11th $\\{$USENIX$\\}$ Symposium on Operating Systems Design and Implementation ($\\{$OSDI$\\}$ 14)_. 583–598.
  * Li et al. (2020) Pengcheng Li, Runze Li, Qing Da, An-Xiang Zeng, and Lijun Zhang. 2020.  Improving Multi-Scenario Learning to Rank in E-commerce by Exploiting Task Relationships in the Label Space. In _Proceedings of the 29th ACM International Conference on Information & Knowledge Management_. 2605–2612.
  * Li et al. (2018a) Youjie Li, Mingchao Yu, Songze Li, Salman Avestimehr, Nam Sung Kim, and Alexander Schwing. 2018a.  Pipe-SGD: a decentralized pipelined SGD framework for distributed deep net training. In _Proceedings of the 32nd International Conference on Neural Information Processing Systems_. 8056–8067.
  * Li et al. (2018b) Zhi Li, Hongke Zhao, Qi Liu, Zhenya Huang, Tao Mei, and Enhong Chen. 2018b.  Learning from history and present: Next-item recommendation via discriminatively exploiting user behaviors. In _Proceedings of the 24th ACM SIGKDD International Conference on Knowledge Discovery & Data Mining_. 1734–1743.
  * Li et al. (2021) Zhuohan Li, Siyuan Zhuang, Shiyuan Guo, Danyang Zhuo, Hao Zhang, Dawn Song, and Ion Stoica. 2021.  TeraPipe: Token-Level Pipeline Parallelism for Training Large-Scale Language Models.  _arXiv preprint arXiv:2102.07988_ (2021).
  * Lian et al. (2015) Xiangru Lian, Yijun Huang, Yuncheng Li, and Ji Liu. 2015\.  Asynchronous parallel stochastic gradient for nonconvex optimization.  _Advances in Neural Information Processing Systems_ 28 (2015), 2737–2745.
  * Lian et al. (2021) Xiangru Lian, Binhang Yuan, Xuefeng Zhu, Yulong Wang, Yongjun He, Honghuan Wu, Lei Sun, Haodong Lyu, Chengjun Liu, Xing Dong, et al. 2021\.  Persia: A Hybrid System Scaling Deep Learning Based Recommenders up to 100 Trillion Parameters.  _arXiv preprint arXiv:2111.05897_ (2021).
  * Lian et al. (2017) Xiangru Lian, Ce Zhang, Huan Zhang, Cho-Jui Hsieh, Wei Zhang, and Ji Liu. 2017\.  Can decentralized algorithms outperform centralized algorithms? a case study for decentralized parallel stochastic gradient descent. In _Proceedings of the 31st International Conference on Neural Information Processing Systems_. 5336–5346.
  * Lian et al. (2016) Xiangru Lian, Huan Zhang, Cho-Jui Hsieh, Yijun Huang, and Ji Liu. 2016.  A Comprehensive Linear Speedup Analysis for Asynchronous Stochastic Parallel Optimization from Zeroth-Order to First-Order.  _NIPS_ (2016).
  * Lian et al. (2018) Xiangru Lian, Wei Zhang, Ce Zhang, and Ji Liu. 2018\.  Asynchronous decentralized parallel stochastic gradient descent. In _International Conference on Machine Learning_. PMLR, 3043–3052.
  * Lin et al. (2019) Tao Lin, Sebastian U Stich, Kumar Kshitij Patel, and Martin Jaggi. 2019.  Don’t Use Large Mini-batches, Use Local SGD. In _International Conference on Learning Representations_.
  * Liu et al. (2017) David C Liu, Stephanie Rogers, Raymond Shiau, Dmitry Kislyuk, Kevin C Ma, Zhigang Zhong, Jenny Liu, and Yushi Jing. 2017\.  Related pins at pinterest: The evolution of a real-world recommender system. In _Proceedings of the 26th international conference on world wide web companion_. 583–592.
  * Liu and Zhang (2020) Ji Liu and Ce Zhang. 2020\.  Distributed Learning Systems with First-Order Methods.  _Foundations and Trends® in Databases_ 9, 1 (2020), 1–100.
  * Lu et al. (2018) Yichao Lu, Ruihai Dong, and Barry Smyth. 2018.  Why I like it: multi-task learning for recommendation and explanation. In _Proceedings of the 12th ACM Conference on Recommender Systems_. 4–12.
  * Miao et al. (2022) Xupeng Miao, Hailin Zhang, Yining Shi, Xiaonan Nie, Zhi Yang, Yangyu Tao, and Bin Cui. 2022.  HET: Scaling out Huge Embedding Model Training via Cache-enabled Distributed Framework.  _Proc. VLDB Endow._ https://doi.org/10.14778/3489496.3489511
  * Mudigere et al. (2021) Dheevatsa Mudigere, Yuchen Hao, Jianyu Huang, Andrew Tulloch, Srinivas Sridharan, Xing Liu, Mustafa Ozdal, Jade Nie, Jongsoo Park, Liang Luo, et al. 2021\.  High-performance, distributed training of large-scale deep learning recommendation models.  _arXiv preprint arXiv:2104.05158_ (2021).
  * Naumov et al. (2020) Maxim Naumov, John Kim, Dheevatsa Mudigere, Srinivas Sridharan, Xiaodong Wang, Whitney Zhao, Serhat Yilmaz, Changkyu Kim, Hector Yuen, Mustafa Ozdal, et al. 2020\.  Deep learning training in facebook data centers: Design of scale-up and scale-out systems.  _arXiv preprint arXiv:2003.09518_ (2020).
  * Nguyen et al. (2018) Lam Nguyen, Phuong Ha Nguyen, Marten Dijk, Peter Richtárik, Katya Scheinberg, and Martin Takác. 2018.  SGD and Hogwild! convergence without the bounded gradients assumption. In _International Conference on Machine Learning_. PMLR, 3750–3758.
  * Niu et al. (2011) Feng Niu, Benjamin Recht, Christopher Re, and Stephen J Wright. 2011\.  HOGWILD! a lock-free approach to parallelizing stochastic gradient descent. In _Proceedings of the 24th International Conference on Neural Information Processing Systems_. 693–701.
  * Pan et al. (2020) Heng Pan, Zhenyu Li, JianBo Dong, Zheng Cao, Tao Lan, Di Zhang, Gareth Tyson, and Gaogang Xie. 2020\.  Dissecting the Communication Latency in Distributed Deep Sparse Learning. In _Proceedings of the ACM Internet Measurement Conference_. 528–534.
  * Park et al. (2020) Jay H Park, Gyeongchan Yun, M Yi Chang, Nguyen T Nguyen, Seungmin Lee, Jaesik Choi, Sam H Noh, and Young-ri Choi. 2020\.  HetPipe: Enabling Large $\\{$DNN$\\}$ Training on (Whimpy) Heterogeneous $\\{$GPU$\\}$ Clusters through Integration of Pipelined Model Parallelism and Data Parallelism. In _2020 $\\{$USENIX$\\}$ Annual Technical Conference ($\\{$USENIX$\\}$$\\{$ATC$\\}$ 20)_. 307–321.
  * Peng et al. (2017) Hao Peng, Shandian Zhe, Xiao Zhang, and Yuan Qi. 2017\.  Asynchronous distributed variational Gaussian process for regression. In _International Conference on Machine Learning_. PMLR, 2788–2797.
  * Rajbhandari et al. (2020) Samyam Rajbhandari, Jeff Rasley, Olatunji Ruwase, and Yuxiong He. 2020.  Zero: Memory optimizations toward training trillion parameter models. In _SC20: International Conference for High Performance Computing, Networking, Storage and Analysis_. IEEE, 1–16.
  * Raman et al. (2019) Parameswaran Raman, Sriram Srinivasan, Shin Matsushima, Xinhua Zhang, Hyokun Yun, and SVN Vishwanathan. 2019.  Scaling multinomial logistic regression via hybrid parallelism. In _Proceedings of the 25th ACM SIGKDD International Conference on Knowledge Discovery & Data Mining_. 1460–1470.
  * Sedhain et al. (2015) Suvash Sedhain, Aditya Krishna Menon, Scott Sanner, and Lexing Xie. 2015.  Autorec: Autoencoders meet collaborative filtering. In _Proceedings of the 24th international conference on World Wide Web_. 111–112.
  * Seo et al. (2017) Sungyong Seo, Jing Huang, Hao Yang, and Yan Liu. 2017\.  Interpretable convolutional neural networks with dual local and global attention for review rating prediction. In _Proceedings of the eleventh ACM conference on recommender systems_. 297–305.
  * Sergeev and Del Balso (2018) Alexander Sergeev and Mike Del Balso. 2018.  Horovod: fast and easy distributed deep learning in TensorFlow.  _arXiv preprint arXiv:1802.05799_ (2018).
  * Shazeer et al. (2018) Noam Shazeer, Youlong Cheng, Niki Parmar, Dustin Tran, Ashish Vaswani, Penporn Koanantakool, Peter Hawkins, HyoukJoong Lee, Mingsheng Hong, Cliff Young, et al. 2018\.  Mesh-TensorFlow: Deep Learning for Supercomputers.  _Advances in Neural Information Processing Systems_ 31 (2018), 10414–10423.
  * Shoeybi et al. (2019) Mohammad Shoeybi, Mostofa Patwary, Raul Puri, Patrick LeGresley, Jared Casper, and Bryan Catanzaro. 2019\.  Megatron-lm: Training multi-billion parameter language models using model parallelism.  _arXiv preprint arXiv:1909.08053_ (2019).
  * Simsekli et al. (2018) Umut Simsekli, Cagatay Yildiz, Than Huy Nguyen, Taylan Cemgil, and Gael Richard. 2018.  Asynchronous stochastic quasi-Newton MCMC for non-convex optimization. In _International Conference on Machine Learning_. PMLR, 4674–4683.
  * Stich (2018) Sebastian U Stich. 2018\.  Local SGD Converges Fast and Communicates Little. In _International Conference on Learning Representations_.
  * Su et al. (2017) Yuhan Su, Zhongming Jin, Ying Chen, Xinghai Sun, Yaming Yang, Fangzheng Qiao, Fen Xia, and Wei Xu. 2017\.  Improving click-through rate prediction accuracy in online advertising by transfer learning. In _Proceedings of the International Conference on Web Intelligence_. 1018–1025.
  * Tan et al. (2016) Yong Kiam Tan, Xinxing Xu, and Yong Liu. 2016.  Improved recurrent neural networks for session-based recommendations. In _Proceedings of the 1st workshop on deep learning for recommender systems_. 17–22.
  * Tang et al. (2019a) Hanlin Tang, Xiangru Lian, Shuang Qiu, Lei Yuan, Ce Zhang, Tong Zhang, and Ji Liu. 2019a.  Deepsqueeze: Parallel stochastic gradient descent with double-pass error-compensated compression.  _arXiv preprint arXiv:1907.07346_ (2019).
  * Tang et al. (2018) Hanlin Tang, Xiangru Lian, Ming Yan, Ce Zhang, and Ji Liu. 2018.  D2: Decentralized training over decentralized data. In _International Conference on Machine Learning_. PMLR, 4848–4856.
  * Tang et al. (2019b) Hanlin Tang, Chen Yu, Xiangru Lian, Tong Zhang, and Ji Liu. 2019b.  Doublesqueeze: Parallel stochastic gradient descent with double-pass error-compensated compression. In _International Conference on Machine Learning_. PMLR, 6155–6165.
  * Tang and Wang (2018) Jiaxi Tang and Ke Wang. 2018.  Personalized top-n sequential recommendation via convolutional sequence embedding. In _Proceedings of the Eleventh ACM International Conference on Web Search and Data Mining_. 565–573.
  * Tarnawski et al. (2020) Jakub M Tarnawski, Amar Phanishayee, Nikhil Devanur, Divya Mahajan, and Fanny Nina Paravecino. 2020\.  Efficient Algorithms for Device Placement of DNN Graph Operators.  _Advances in Neural Information Processing Systems_ 33 (2020).
  * Tuan and Phuong (2017) Trinh Xuan Tuan and Tu Minh Phuong. 2017.  3D convolutional networks for session-based recommendation with content features. In _Proceedings of the eleventh ACM conference on recommender systems_. 138–146.
  * Twardowski (2016) Bartłomiej Twardowski. 2016\.  Modelling contextual information in session-aware recommender systems with neural networks. In _Proceedings of the 10th ACM Conference on Recommender Systems_. 273–276.
  * Van Den Oord et al. (2013) Aäron Van Den Oord, Sander Dieleman, and Benjamin Schrauwen. 2013.  Deep content-based music recommendation. In _Neural Information Processing Systems Conference (NIPS 2013)_ , Vol. 26. Neural Information Processing Systems Foundation (NIPS).
  * Wang et al. (2018b) Hongyi Wang, Scott Sievert, Zachary Charles, Shengchao Liu, Stephen Wright, and Dimitris Papailiopoulos. 2018b.  ATOMO: communication-efficient learning via atomic sparsification. In _Proceedings of the 32nd International Conference on Neural Information Processing Systems_. 9872–9883.
  * Wang et al. (2018a) Jizhe Wang, Pipei Huang, Huan Zhao, Zhibo Zhang, Binqiang Zhao, and Dik Lun Lee. 2018a.  Billion-scale commodity embedding for e-commerce recommendation in alibaba. In _Proceedings of the 24th ACM SIGKDD International Conference on Knowledge Discovery & Data Mining_. 839–848.
  * Wang and Joshi (2019) Jianyu Wang and Gauri Joshi. 2019.  Adaptive Communication Strategies to Achieve the Best Error-Runtime Trade-off in Local-update SGD. In _Systems and Machine Learning (SysML) Conference_.
  * Wang et al. (2017a) Jialei Wang, Mladen Kolar, Nathan Srebro, and Tong Zhang. 2017a.  Efficient distributed learning with sparsity. In _International Conference on Machine Learning_. PMLR, 3636–3645.
  * Wang et al. (2019) Minjie Wang, Chien-chin Huang, and Jinyang Li. 2019.  Supporting very large models using automatic dataflow graph partitioning. In _Proceedings of the Fourteenth EuroSys Conference 2019_. 1–17.
  * Wang et al. (2018c) Nan Wang, Hongning Wang, Yiling Jia, and Yue Yin. 2018c.  Explainable recommendation via multi-task learning in opinionated text data. In _The 41st International ACM SIGIR Conference on Research & Development in Information Retrieval_. 165–174.
  * Wang et al. (2016) Shanfeng Wang, Maoguo Gong, Haoliang Li, and Junwei Yang. 2016\.  Multi-objective optimization for long tail recommendation.  _Knowledge-Based Systems_ 104 (2016), 145–155.
  * Wang et al. (2017b) Suhang Wang, Yilin Wang, Jiliang Tang, Kai Shu, Suhas Ranganath, and Huan Liu. 2017b.  What your images reveal: Exploiting visual contents for point-of-interest recommendation. In _Proceedings of the 26th international conference on world wide web_. 391–400.
  * Wang and Wang (2014) Xinxi Wang and Ye Wang. 2014.  Improving content-based and hybrid music recommendation using deep learning. In _Proceedings of the 22nd ACM international conference on Multimedia_. 627–636.
  * Wangni et al. (2018) J Wangni, J Liu, J Wang, and T Zhang. 2018\.  Gradient sparsification for communication-efficient distributed optimization.  _Advances in Neural Information Processing Systems_ 31 (2018), 1299\.
  * Wen et al. (2018) Jiqing Wen, James She, Xiaopeng Li, and Hui Mao. 2018\.  Visual background recommendation for dance performances using deep matrix factorization.  _ACM Transactions on Multimedia Computing, Communications, and Applications (TOMM)_ 14, 1 (2018), 1–19.
  * Wen et al. (2017) Wei Wen, Cong Xu, Feng Yan, Chunpeng Wu, Yandan Wang, Yiran Chen, and Hai Li. 2017.  TernGrad: ternary gradients to reduce communication in distributed deep learning. In _Proceedings of the 31st International Conference on Neural Information Processing Systems_. 1508–1518.
  * Wilkening et al. (2021) Mark Wilkening, Udit Gupta, Samuel Hsia, Caroline Trippel, Carole-Jean Wu, David Brooks, and Gu-Yeon Wei. 2021.  RecSSD: Near Data Processing for Solid State Drive Based Recommendation Inference.  _arXiv preprint arXiv:2102.00075_ (2021).
  * Wu et al. (2016b) Caihua Wu, Junwei Wang, Juntao Liu, and Wenyu Liu. 2016b.  Recurrent neural network based recommendation for time heterogeneous feedback.  _Knowledge-Based Systems_ 109 (2016), 90–103.
  * Wu et al. (2016a) Sai Wu, Weichao Ren, Chengchao Yu, Gang Chen, Dongxiang Zhang, and Jingbo Zhu. 2016a.  Personal recommendation using deep recurrent neural networks in NetEase. In _2016 IEEE 32nd international conference on data engineering (ICDE)_. IEEE, 1218–1229.
  * Xie et al. (2020) Minhui Xie, Kai Ren, Youyou Lu, Guangxu Yang, Qingxing Xu, Bihai Wu, Jiazhen Lin, Hongbo Ao, Wanhong Xu, and Jiwu Shu. 2020\.  Kraken: memory-efficient continual learning for large-scale real-time recommendations. In _SC20: International Conference for High Performance Computing, Networking, Storage and Analysis_. IEEE, 1–17.
  * Xie et al. (2018) Xing Xie, Jianxun Lian, Zheng Liu, Xiting Wang, Fangzhao Wu, Hongwei Wang, and Zhongxia Chen. 2018.  Personalized recommendation systems: Five hot research topics you must know.  _Microsoft Research Lab-Asia_ (2018).
  * Xu et al. (2016) Zhenghua Xu, Cheng Chen, Thomas Lukasiewicz, Yishu Miao, and Xiangwu Meng. 2016.  Tag-aware personalized recommendation using a deep-semantic similarity model with negative sampling. In _Proceedings of the 25th ACM International on Conference on Information and Knowledge Management_. 1921–1924.
  * Yang et al. (2017) Carl Yang, Lanxiao Bai, Chao Zhang, Quan Yuan, and Jiawei Han. 2017.  Bridging collaborative filtering and semi-supervised learning: a neural approach for poi recommendation. In _Proceedings of the 23rd ACM SIGKDD International Conference on Knowledge Discovery and Data Mining_. 1245–1254.
  * Zhang et al. (2020) Caojin Zhang, Yicun Liu, Yuanpu Xie, Sofia Ira Ktena, Alykhan Tejani, Akshay Gupta, Pranay Kumar Myana, Deepak Dilipkumar, Suvadip Paul, Ikuhiro Ihara, et al. 2020\.  Model Size Reduction Using Frequency Based Double Hashing for Recommender Systems. In _Fourteenth ACM Conference on Recommender Systems_. 521–526.
  * Zhang et al. (2017b) Hantian Zhang, Jerry Li, Kaan Kara, Dan Alistarh, Ji Liu, and Ce Zhang. 2017b.  Zipml: Training linear models with end-to-end low precision, and a little bit of deep learning. In _International Conference on Machine Learning_. PMLR, 4035–4043.
  * Zhang et al. (2017c) Qi Zhang, Jiawen Wang, Haoran Huang, Xuanjing Huang, and Yeyun Gong. 2017c.  Hashtag Recommendation for Multimodal Microblog Using Co-Attention Network.. In _IJCAI_. 3420–3426.
  * Zhang et al. (2019c) Shuai Zhang, Lina Yao, Aixin Sun, and Yi Tay. 2019c.  Deep learning based recommender system: A survey and new perspectives.  _ACM Computing Surveys (CSUR)_ 52, 1 (2019), 1–38.
  * Zhang et al. (2017a) Yongfeng Zhang, Qingyao Ai, Xu Chen, and W Bruce Croft. 2017a.  Joint representation learning for top-n recommendation with heterogeneous information sources. In _Proceedings of the 2017 ACM on Conference on Information and Knowledge Management_. 1449–1458.
  * Zhang et al. (2019a) Zhipeng Zhang, Bin Cui, Yingxia Shao, Lele Yu, Jiawei Jiang, and Xupeng Miao. 2019a.  Ps2: Parameter server on spark. In _Proceedings of the 2019 ACM SIGMOD International Conference on Management of Data_. 376–388.
  * Zhang et al. (2019b) Zhipeng Zhang, Jiawei Jiang, Wentao Wu, Ce Zhang, Lele Yu, and Bin Cui. 2019b.  Mllib*: Fast training of glms using spark mllib. In _2019 IEEE 35th International Conference on Data Engineering (ICDE)_. IEEE Computer Society, 1778–1789.
  * Zhao et al. (2020b) Weijie Zhao, Deping Xie, Ronglai Jia, Yulei Qian, Ruiquan Ding, Mingming Sun, and Ping Li. 2020b.  Distributed hierarchical gpu parameter server for massive scale deep learning ads systems.  _arXiv preprint arXiv:2003.05622_ (2020).
  * Zhao et al. (2019b) Weijie Zhao, Jingyuan Zhang, Deping Xie, Yulei Qian, Ronglai Jia, and Ping Li. 2019b.  AIBox: CTR prediction model training on a single node. In _Proceedings of the 28th ACM International Conference on Information and Knowledge Management_. 319–328.
  * Zhao et al. (2020a) Xiangyu Zhao, Chong Wang, Ming Chen, Xudong Zheng, Xiaobing Liu, and Jiliang Tang. 2020a.  Autoemb: Automated embedding dimensionality search in streaming recommendations.  _arXiv preprint arXiv:2002.11252_ (2020).
  * Zhao et al. (2018) Xiangyu Zhao, Liang Zhang, Zhuoye Ding, Long Xia, Jiliang Tang, and Dawei Yin. 2018\.  Recommendations with negative feedback via pairwise deep reinforcement learning. In _Proceedings of the 24th ACM SIGKDD International Conference on Knowledge Discovery & Data Mining_. 1040–1048.
  * Zhao et al. (2019a) Zhe Zhao, Lichan Hong, Li Wei, Jilin Chen, Aniruddh Nath, Shawn Andrews, Aditee Kumthekar, Maheswaran Sathiamoorthy, Xinyang Yi, and Ed Chi. 2019a.  Recommending what video to watch next: a multitask ranking system. In _Proceedings of the 13th ACM Conference on Recommender Systems_. 43–51.
  * Zheng et al. (2018) Guanjie Zheng, Fuzheng Zhang, Zihan Zheng, Yang Xiang, Nicholas Jing Yuan, Xing Xie, and Zhenhui Li. 2018.  DRN: A deep reinforcement learning framework for news recommendation. In _Proceedings of the 2018 World Wide Web Conference_. 167–176.
  * Zheng et al. (2017) Shuxin Zheng, Qi Meng, Taifeng Wang, Wei Chen, Nenghai Yu, Zhi-Ming Ma, and Tie-Yan Liu. 2017.  Asynchronous stochastic gradient descent with delay compensation. In _International Conference on Machine Learning_. PMLR, 4120–4129.
  * Zhou et al. (2019) Guorui Zhou, Na Mou, Ying Fan, Qi Pi, Weijie Bian, Chang Zhou, Xiaoqiang Zhu, and Kun Gai. 2019.  Deep interest evolution network for click-through rate prediction. In _Proceedings of the AAAI conference on artificial intelligence_ , Vol. 33. 5941–5948.
  * Zhou et al. (2018b) Guorui Zhou, Xiaoqiang Zhu, Chenru Song, Ying Fan, Han Zhu, Xiao Ma, Yanghui Yan, Junqi Jin, Han Li, and Kun Gai. 2018b.  Deep interest network for click-through rate prediction. In _Proceedings of the 24th ACM SIGKDD International Conference on Knowledge Discovery & Data Mining_. 1059–1068.
  * Zhou et al. (2018a) Zhengyuan Zhou, Panayotis Mertikopoulos, Nicholas Bambos, Peter Glynn, Yinyu Ye, Li-Jia Li, and Li Fei-Fei. 2018a.  Distributed asynchronous optimization with unbounded delays: How slow can you go?. In _International Conference on Machine Learning_. PMLR, 5970–5979.
  * Zhuang et al. (2017) Fuzhen Zhuang, Zhiqiang Zhang, Mingda Qian, Chuan Shi, Xing Xie, and Qing He. 2017\.  Representation learning via Dual-Autoencoder for recommendation.  _Neural Networks_ 90 (2017), 83–89.
  * Zuo et al. (2016) Yi Zuo, Jiulin Zeng, Maoguo Gong, and Licheng Jiao. 2016\.  Tag-aware recommender systems based on deep neural networks.  _Neurocomputing_ 204 (2016), 51–60.

## Appendix

###### Lemma 0.

Under Assumption 1, for any $\delta\in\mathbb{R}^{N^{\mathrm{emb}}}$ the
following inequality holds:

(7)
$\displaystyle\left\|f^{\prime}\left(\mathbf{w}^{\mathrm{emb}}+\delta,\mathbf{w}^{\mathrm{nn}}\right)-f^{\prime}\left(\mathbf{w}^{\mathrm{emb}},\mathbf{w}^{\mathrm{nn}}\right)\right\|\leqslant\alpha
L\left\|\delta\right\|,\quad\forall\mathbf{w}^{\mathrm{emb}}\in\mathbb{R}^{N^{\mathrm{emb}}},\mathbf{w}^{\mathrm{nn}}\in\mathbb{R}^{N^{\mathrm{nn}}}.$

###### Proof.

Define Hessian matrix w.r.t. the sparse weights $\mathbf{w}^{\text{emb}}$:

$\displaystyle\mathbf{H}_{F_{\xi}}^{\text{emb}}:=$
$\displaystyle\left(\begin{array}[]{cccc}\frac{\partial^{2}F\left(\mathbf{w}^{\text{emb}},\mathbf{w}^{\text{nn}};\xi\right)}{\partial\left[\mathbf{w}^{\text{emb}}_{1}\right]^{2}}&\frac{\partial^{2}F\left(\mathbf{w}^{\text{emb}},\mathbf{w}^{\text{nn}};\xi\right)}{\partial\mathbf{w}^{\text{emb}}_{1}\partial\mathbf{w}^{\text{emb}}_{2}}&\cdots&\frac{\partial^{2}F\left(\mathbf{w}^{\text{emb}},\mathbf{w}^{\text{nn}};\xi\right)}{\partial\mathbf{w}^{\text{emb}}_{1}\partial\mathbf{w}^{\text{emb}}_{N^{\text{emb}}}}\\\
\frac{\partial^{2}F\left(\mathbf{w}^{\text{emb}},\mathbf{w}^{\text{nn}};\xi\right)}{\partial\mathbf{w}^{\text{emb}}_{2}\partial\mathbf{w}^{\text{emb}}_{1}}&\frac{\partial^{2}F\left(\mathbf{w}^{\text{emb}},\mathbf{w}^{\text{nn}};\xi\right)}{\partial\left[\mathbf{w}^{\text{emb}}_{2}\right]^{2}}&\cdots&\frac{\partial^{2}F\left(\mathbf{w}^{\text{emb}},\mathbf{w}^{\text{nn}};\xi\right)}{\partial\mathbf{w}^{\text{emb}}_{2}\partial\mathbf{w}^{\text{emb}}_{N^{\text{emb}}}}\\\
\vdots&\vdots&\ddots&\vdots\\\
\frac{\partial^{2}F\left(\mathbf{w}^{\text{emb}},\mathbf{w}^{\text{nn}};\xi\right)}{\partial\mathbf{w}^{\text{emb}}_{n}\partial\mathbf{w}^{\text{emb}}_{1}}&\frac{\partial^{2}F\left(\mathbf{w}^{\text{emb}},\mathbf{w}^{\text{nn}};\xi\right)}{\partial\mathbf{w}^{\text{emb}}_{n}\partial\mathbf{w}^{\text{emb}}_{2}}&\cdots&\frac{\partial^{2}F\left(\mathbf{w}^{\text{emb}},\mathbf{w}^{\text{nn}};\xi\right)}{\partial\left[\mathbf{w}^{\text{emb}}_{n}\right]^{2}}\end{array}\right),$
$\displaystyle\mathbf{H}_{f}^{\text{emb}}:=$
$\displaystyle\left(\begin{array}[]{cccc}\frac{\partial^{2}f\left(\mathbf{w}^{\text{emb}},\mathbf{w}^{\text{nn}}\right)}{\partial\left[\mathbf{w}^{\text{emb}}_{1}\right]^{2}}&\frac{\partial^{2}f\left(\mathbf{w}^{\text{emb}},\mathbf{w}^{\text{nn}}\right)}{\partial\mathbf{w}^{\text{emb}}_{1}\partial\mathbf{w}^{\text{emb}}_{2}}&\cdots&\frac{\partial^{2}f\left(\mathbf{w}^{\text{emb}},\mathbf{w}^{\text{nn}}\right)}{\partial\mathbf{w}^{\text{emb}}_{1}\partial\mathbf{w}^{\text{emb}}_{N^{\text{emb}}}}\\\
\frac{\partial^{2}f\left(\mathbf{w}^{\text{emb}},\mathbf{w}^{\text{nn}}\right)}{\partial\mathbf{w}^{\text{emb}}_{2}\partial\mathbf{w}^{\text{emb}}_{1}}&\frac{\partial^{2}f\left(\mathbf{w}^{\text{emb}},\mathbf{w}^{\text{nn}}\right)}{\partial\left[\mathbf{w}^{\text{emb}}_{2}\right]^{2}}&\cdots&\frac{\partial^{2}f\left(\mathbf{w}^{\text{emb}},\mathbf{w}^{\text{nn}}\right)}{\partial\mathbf{w}^{\text{emb}}_{2}\partial\mathbf{w}^{\text{emb}}_{N^{\text{emb}}}}\\\
\vdots&\vdots&\ddots&\vdots\\\
\frac{\partial^{2}f\left(\mathbf{w}^{\text{emb}},\mathbf{w}^{\text{nn}}\right)}{\partial\mathbf{w}^{\text{emb}}_{n}\partial\mathbf{w}^{\text{emb}}_{1}}&\frac{\partial^{2}f\left(\mathbf{w}^{\text{emb}},\mathbf{w}^{\text{nn}}\right)}{\partial\mathbf{w}^{\text{emb}}_{n}\partial\mathbf{w}^{\text{emb}}_{2}}&\cdots&\frac{\partial^{2}f\left(\mathbf{w}^{\text{emb}},\mathbf{w}^{\text{nn}}\right)}{\partial\left[\mathbf{w}^{\text{emb}}_{n}\right]^{2}}\end{array}\right).$

From the Lipschitzian assumption (3) and the stochastic gradient function
$F^{\prime}(\cdot;\xi)$ is differentiable, we obtain the Hessian matrix of
$F(\cdot;\xi)$ is $\preccurlyeq LI$, where $I$ is the identity matrix. Noting
that since for each sample, the $\mathbf{w}^{\text{emb}}$ is used by a lookup
operation (LABEL:eq:activation-set-def) where only a subset of the weights get
used. For any element $\mathbf{w}^{\text{emb}}_{i}$, if
$i\not\in\Omega_{\xi}$, $\mathbf{w}^{\text{emb}}_{i}$ will have no effect on
the value of $F^{\prime}(\cdot;\xi)$. This implies the corresponding elements
are zero in $F(\cdot;\xi)$’s Hessian matrix, and the following inequality
holds:

$\displaystyle\mathbf{H}_{F_{\xi}}^{\text{emb}}\preccurlyeq$ $\displaystyle
LI_{\xi},$

where

$\displaystyle(I_{\xi})_{i,j}=$
$\displaystyle\left\\{\begin{array}[]{ll}1,&\text{if }i=j\text{ and
}i\in\Omega_{\xi},\\\ 0,&\text{otherwise.}\end{array}\right..$

Note that from the definition of $f$ in (1), we obtain the Hessian of $f$ is
the expectation of the Hessian of $F$:

$\displaystyle\mathbf{H}_{f}^{\text{emb}}=$
$\displaystyle\mathbb{E}_{\xi}\mathbf{H}_{F_{\xi}}^{\text{emb}}\preccurlyeq
L\mathbb{E}_{\xi}I_{\xi}.$

Finally following the definition of $\alpha$, we obtain
$\mathbf{H}_{f}^{\text{emb}}\preccurlyeq L\mathbb{E}_{\xi}I_{\xi}\preccurlyeq
L\alpha$, which leads to the $\alpha L$ Lipschitzian continuity in the sparse
weights (7). ∎

Proof to Theorem 1

###### Proof.

Since $f$ is defined as the average of $F$, from Lipschitzian gradient
assumption (3), it can be derived that $f$ is also Lipschitzian:

(8)
$\displaystyle\left\|f^{\prime}\left(\mathbf{w}\right)-f^{\prime}\left(\mathbf{w}+\mathbf{\delta}\right)\right\|\leqslant
L\left\|\mathbf{\delta}\right\|,\quad\forall\mathbf{w},\mathbf{\delta}\in\mathbb{R}^{N^{\mathrm{emb}}+N^{\mathrm{nn}}}.$

From the algorithm update rule (2) and the Lipschitzian property (8), we
obtain

$\displaystyle
f\left(\mathbf{w}^{\mathrm{emb}}_{\mathrm{t+1}},\mathbf{w}^{\mathrm{nn}}_{\mathrm{t+1}}\right)-f\left(\mathbf{w}^{\mathrm{emb}}_{\mathrm{t}},\mathbf{w}^{\mathrm{nn}}_{\mathrm{t}}\right)$
$\displaystyle\overset{\eqref{eq:able-snipe-1}}{\leqslant}$
$\displaystyle\left\langle
f^{\prime}\left(\mathbf{w}^{\mathrm{emb}}_{\mathrm{t}},\mathbf{w}^{\mathrm{nn}}_{\mathrm{t}}\right),\mathbf{w}^{\mathrm{emb}}_{\mathrm{t+1}}+\mathbf{w}^{\mathrm{nn}}_{\mathrm{t+1}}-\mathbf{w}^{\mathrm{emb}}_{\mathrm{t}}-\mathbf{w}^{\mathrm{nn}}_{\mathrm{t}}\right\rangle+\frac{L}{2}\left\|\mathbf{w}^{\mathrm{emb}}_{\mathrm{t+1}}+\mathbf{w}^{\mathrm{nn}}_{\mathrm{t+1}}-\mathbf{w}^{\mathrm{emb}}_{\mathrm{t}}-\mathbf{w}^{\mathrm{nn}}_{\mathrm{t}}\right\|^{2}$
$\displaystyle\overset{\eqref{eq:59f6c90b}}{=}$
$\displaystyle-\gamma\left\langle
f^{\prime}\left(\mathbf{w}^{\mathrm{emb}}_{\mathrm{t}},\mathbf{w}^{\mathrm{nn}}_{\mathrm{t}}\right),{{F^{\mathrm{emb}}_{t}}}^{\prime}+{{F^{\mathrm{nn}}_{t}}}^{\prime}\right\rangle+\frac{L\gamma^{2}}{2}\left\|{{F^{\mathrm{emb}}_{t}}}^{\prime}+{{F^{\mathrm{nn}}_{t}}}^{\prime}\right\|^{2}$
$\displaystyle=$ $\displaystyle-\gamma\left\langle
f^{\prime}\left(\mathbf{w}^{\mathrm{emb}}_{\mathrm{t}},\mathbf{w}^{\mathrm{nn}}_{\mathrm{t}}\right),F_{t}^{\prime}\right\rangle+\frac{L\gamma^{2}}{2}\left\|F_{t}^{\prime}\right\|^{2}.$

Taking the expectation on both sides, we obtain

(9)
$\displaystyle\mathbb{E}\left[f\left(\mathbf{w}^{\mathrm{emb}}_{\mathrm{t+1}},\mathbf{w}^{\mathrm{nn}}_{\mathrm{t+1}}\right)-f\left(\mathbf{w}^{\mathrm{emb}}_{\mathrm{t}},\mathbf{w}^{\mathrm{nn}}_{\mathrm{t}}\right)\right]\leqslant-\gamma\underbrace{\mathbb{E}\left[\left\langle
f^{\prime}\left(\mathbf{w}^{\mathrm{emb}}_{\mathrm{t}},\mathbf{w}^{\mathrm{nn}}_{\mathrm{t}}\right),F_{t}^{\prime}\right\rangle\right]}_{=:T_{1}}+\frac{L\gamma^{2}}{2}\underbrace{\mathbb{E}\left[\left\|F_{t}^{\prime}\right\|^{2}\right]}_{=:T_{2}}.$

We show the bound for the terms $T_{1}$ and $T_{2}$ respectively, in the
following paragraphs. For $T_{2}$, from the unbiased property of stochastic
gradients (1) and the bounded variance assumption (4) we obtain

$\displaystyle T_{2}=$
$\displaystyle\mathbb{E}\left[\left\|F_{t}^{\prime}\right\|^{2}\right]$
$\displaystyle=$
$\displaystyle\mathbb{E}\left[\left\|F_{t}^{\prime}-f^{\prime}\left(\mathbf{w}^{\mathrm{emb}}_{\mathrm{\mathrm{D}(t)}},\mathbf{w}^{\mathrm{nn}}_{\mathrm{t}}\right)+f^{\prime}\left(\mathbf{w}^{\mathrm{emb}}_{\mathrm{\mathrm{D}(t)}},\mathbf{w}^{\mathrm{nn}}_{\mathrm{t}}\right)\right\|^{2}\right]$
$\displaystyle=$
$\displaystyle\mathbb{E}\left[\left\|F_{t}^{\prime}-f^{\prime}\left(\mathbf{w}^{\mathrm{emb}}_{\mathrm{\mathrm{D}(t)}},\mathbf{w}^{\mathrm{nn}}_{\mathrm{t}}\right)\right\|^{2}+\left\|f^{\prime}\left(\mathbf{w}^{\mathrm{emb}}_{\mathrm{\mathrm{D}(t)}},\mathbf{w}^{\mathrm{nn}}_{\mathrm{t}}\right)\right\|^{2}\right]+2\underbrace{\mathbb{E}\left[\left\langle
F_{t}^{\prime}-f^{\prime}\left(\mathbf{w}^{\mathrm{emb}}_{\mathrm{\mathrm{D}(t)}},\mathbf{w}^{\mathrm{nn}}_{\mathrm{t}}\right),f^{\prime}\left(\mathbf{w}^{\mathrm{emb}}_{\mathrm{\mathrm{D}(t)}},\mathbf{w}^{\mathrm{nn}}_{\mathrm{t}}\right)\right\rangle\right]}_{=0,\text{
from \eqref{eq:deciding-dory}}}$ (10)
$\displaystyle\overset{\eqref{eq:credible-bird}}{\leqslant}$
$\displaystyle\sigma^{2}+\mathbb{E}\left[\left\|f^{\prime}\left(\mathbf{w}^{\mathrm{emb}}_{\mathrm{\mathrm{D}(t)}},\mathbf{w}^{\mathrm{nn}}_{\mathrm{t}}\right)\right\|^{2}\right].$

For $T_{1}$, we have

$\displaystyle T_{1}=$ $\displaystyle\mathbb{E}\left[\left\langle
f^{\prime}\left(\mathbf{w}^{\mathrm{emb}}_{\mathrm{t}},\mathbf{w}^{\mathrm{nn}}_{\mathrm{t}}\right),F_{t}^{\prime}\right\rangle\right]$
$\displaystyle\overset{\eqref{eq:deciding-dory}}{=}$
$\displaystyle\mathbb{E}\left[\left\langle
f^{\prime}(\mathbf{w}^{\mathrm{emb}}_{\mathrm{t}},\mathbf{w}^{\mathrm{nn}}_{\mathrm{t}}),f^{\prime}\left(\mathbf{w}^{\mathrm{emb}}_{\mathrm{\mathrm{D}(t)}},\mathbf{w}^{\mathrm{nn}}_{\mathrm{t}}\right)\right\rangle\right]$
(11) $\displaystyle=$
$\displaystyle\frac{1}{2}\mathbb{E}\left[\left\|f^{\prime}\left(\mathbf{w}^{\mathrm{emb}}_{\mathrm{t}},\mathbf{w}^{\mathrm{nn}}_{\mathrm{t}}\right)\right\|^{2}+\left\|f^{\prime}\left(\mathbf{w}^{\mathrm{emb}}_{\mathrm{\mathrm{D}(t)}},\mathbf{w}^{\mathrm{nn}}_{\mathrm{t}}\right)\right\|^{2}-\left\|f^{\prime}\left(\mathbf{w}^{\mathrm{emb}}_{\mathrm{t}},\mathbf{w}^{\mathrm{nn}}_{\mathrm{t}}\right)-f^{\prime}\left(\mathbf{w}^{\mathrm{emb}}_{\mathrm{\mathrm{D}(t)}},\mathbf{w}^{\mathrm{nn}}_{\mathrm{t}}\right)\right\|^{2}\right],$

where the last step comes from the fact that for any $\mathbf{a},\mathbf{b}$
the following equality holds:
$\left\langle\mathbf{a},\mathbf{b}\right\rangle=\frac{1}{2}\left\|\mathbf{a}-\mathbf{b}\right\|^{2}-\frac{1}{2}\left\|\mathbf{a}\right\|^{2}-\frac{1}{2}\left\|\mathbf{b}\right\|^{2}$.

Using the bounds for $T_{1}$ (11) and $T_{2}$ (10) in (9), we obtain

$\displaystyle\mathbb{E}\left[f\left(\mathbf{w}^{\mathrm{emb}}_{\mathrm{t+1}},\mathbf{w}^{\mathrm{nn}}_{\mathrm{t+1}}\right)-f\left(\mathbf{w}^{\mathrm{emb}}_{\mathrm{t}},\mathbf{w}^{\mathrm{nn}}_{\mathrm{t}}\right)\right]$
$\displaystyle\leqslant$
$\displaystyle-\frac{\gamma}{2}\mathbb{E}\left[\left\|f^{\prime}\left(\mathbf{w}^{\mathrm{emb}}_{\mathrm{t}},\mathbf{w}^{\mathrm{nn}}_{\mathrm{t}}\right)\right\|^{2}+\left\|f^{\prime}\left(\mathbf{w}^{\mathrm{emb}}_{\mathrm{\mathrm{D}(t)}},\mathbf{w}^{\mathrm{nn}}_{\mathrm{t}}\right)\right\|^{2}-\left\|f^{\prime}\left(\mathbf{w}^{\mathrm{emb}}_{\mathrm{t}},\mathbf{w}^{\mathrm{nn}}_{\mathrm{t}}\right)-f^{\prime}\left(\mathbf{w}^{\mathrm{emb}}_{\mathrm{\mathrm{D}(t)}},\mathbf{w}^{\mathrm{nn}}_{\mathrm{t}}]\right)\right\|^{2}\right]$
$\displaystyle+\frac{L\gamma^{2}}{2}\mathbb{E}\left[\left\|f^{\prime}\left(\mathbf{w}^{\mathrm{emb}}_{\mathrm{\mathrm{D}(t)}},\mathbf{w}^{\mathrm{nn}}_{\mathrm{t}}\right)\right\|^{2}\right]+\frac{L\gamma^{2}}{2}\sigma^{2}$
$\displaystyle=$
$\displaystyle-\frac{\gamma}{2}\mathbb{E}\left[\left\|f^{\prime}\left(\mathbf{w}^{\mathrm{emb}}_{\mathrm{t}},\mathbf{w}^{\mathrm{nn}}_{\mathrm{t}}\right)\right\|^{2}\right]-\frac{\gamma}{2}\left(1-\gamma
L\right)\mathbb{E}\left[\left\|f^{\prime}\left(\mathbf{w}^{\mathrm{emb}}_{\mathrm{\mathrm{D}(t)}},\mathbf{w}^{\mathrm{nn}}_{\mathrm{t}}\right)\right\|^{2}\right]+\frac{L\gamma^{2}}{2}\sigma^{2}+\frac{\gamma}{2}\mathbb{E}\left[\left\|f^{\prime}\left(\mathbf{w}^{\mathrm{emb}}_{\mathrm{t}},\mathbf{w}^{\mathrm{nn}}_{\mathrm{t}}\right)-f^{\prime}\left(\mathbf{w}^{\mathrm{emb}}_{\mathrm{\mathrm{D}(t)}},\mathbf{w}^{\mathrm{nn}}_{\mathrm{t}}\right)\right\|^{2}\right]$
(12) $\displaystyle\leqslant$
$\displaystyle-\frac{\gamma}{2}\mathbb{E}\left[\left\|f^{\prime}\left(\mathbf{w}^{\mathrm{emb}}_{\mathrm{t}},\mathbf{w}^{\mathrm{nn}}_{\mathrm{t}}\right)\right\|^{2}\right]+\frac{L\gamma^{2}}{2}\sigma^{2}+\frac{\gamma}{2}\underbrace{\mathbb{E}\left[\left\|f^{\prime}\left(\mathbf{w}^{\mathrm{emb}}_{\mathrm{t}},\mathbf{w}^{\mathrm{nn}}_{\mathrm{t}}\right)-f^{\prime}\left(\mathbf{w}^{\mathrm{emb}}_{\mathrm{\mathrm{D}(t)}},\mathbf{w}^{\mathrm{nn}}_{\mathrm{t}}\right)\right\|^{2}\right]}_{=:\omega_{t}},$

where the last inequality is obtained by choosing a sufficiently small
learning rate $\gamma$ satisfying $1\geqslant\gamma L$.

We then bound $\omega_{t}$ in order to get the final convergence rate result.
Denoting

$\displaystyle L_{\text{emb}}:=$
$\displaystyle\min\left\\{1,\alpha\right\\}L,$
$\displaystyle{{F^{\mathrm{emb}}_{t}}}^{\prime}(t):=$
$\displaystyle\nabla_{\mathbf{w}^{\mathrm{emb}}}f\left(\mathbf{w}^{\mathrm{emb}}_{\mathrm{\mathrm{D}(t)}},\mathbf{w}^{\mathrm{emb}}_{\mathrm{t}}\right),$
$\displaystyle{{F^{\mathrm{nn}}_{t}}}^{\prime}(t):=$
$\displaystyle\nabla_{\mathbf{w}^{\mathrm{nn}}}f\left(\mathbf{w}^{\mathrm{emb}}_{\mathrm{\mathrm{D}(t)}},\mathbf{w}^{\mathrm{emb}}_{\mathrm{t}}\right),$

it follows from Lemma 1 that

$\displaystyle\omega_{t}=$
$\displaystyle\mathbb{E}\left[\left\|f^{\prime}\left(\mathbf{w}^{\mathrm{emb}}_{\mathrm{t}},\mathbf{w}^{\mathrm{nn}}_{\mathrm{t}}\right)-f^{\prime}\left(\mathbf{w}^{\mathrm{emb}}_{\mathrm{\mathrm{D}(t)}},\mathbf{w}^{\mathrm{nn}}_{\mathrm{t}}\right)\right\|^{2}\right]$
$\displaystyle\overset{\eqref{eq:crack-skylark-1}}{\leqslant}$ $\displaystyle
L^{2}_{\text{emb}}\mathbb{E}\left[\left\|\mathbf{w}^{\mathrm{emb}}_{\mathrm{t}}-\mathbf{w}^{\mathrm{emb}}_{\mathrm{\mathrm{D}(t)}}\right\|^{2}\right]$
$\displaystyle\overset{\eqref{eq:59f6c90b}}{=}$
$\displaystyle\gamma^{2}L^{2}_{\text{emb}}\mathbb{E}\left[\left\|\sum^{t-1}_{k=\mathrm{D}(t)}{{F^{\mathrm{emb}}_{k}}}^{\prime}\right\|^{2}\right]$
$\displaystyle=$
$\displaystyle\gamma^{2}L^{2}_{\text{emb}}\mathbb{E}\left[\left\|\sum^{t-1}_{k=\mathrm{D}(t)}{{F^{\mathrm{emb}}_{k}}}^{\prime}-\sum^{t-1}_{k=\mathrm{D}(t)}{{F^{\mathrm{emb}}_{k}}}^{\prime}+\sum^{t-1}_{k=\mathrm{D}(t)}{{F^{\mathrm{emb}}_{k}}}^{\prime}\right\|^{2}\right]$
$\displaystyle=$
$\displaystyle\gamma^{2}L^{2}_{\text{emb}}\mathbb{E}\left[\left\|\sum^{t-1}_{k=\mathrm{D}(t)}{{F^{\mathrm{emb}}_{k}}}^{\prime}-\sum^{t-1}_{k=\mathrm{D}(t)}{{F^{\mathrm{emb}}_{k}}}^{\prime}\right\|^{2}\right]+\gamma^{2}L^{2}_{\text{emb}}\mathbb{E}\left[\left\|\sum^{t-1}_{k=\mathrm{D}(t)}{{F^{\mathrm{emb}}_{k}}}^{\prime}\right\|^{2}\right]$
$\displaystyle\overset{\eqref{eq:credible-bird}}{\leqslant}$
$\displaystyle\gamma^{2}L^{2}_{\text{emb}}\tau\sigma^{2}+\gamma^{2}L^{2}_{\text{emb}}\mathbb{E}\left[\left\|\sum^{t-1}_{k=\mathrm{D}(t)}{{F^{\mathrm{emb}}_{k}}}^{\prime}\right\|^{2}\right]$
(13) $\displaystyle\leqslant$ $\displaystyle\gamma^{2}\tau
L^{2}_{\text{emb}}\tau\sigma^{2}+\gamma^{2}L^{2}_{\text{emb}}\mathbb{E}\left[\sum^{t-1}_{k=\mathrm{D}(t)}\left\|{{F^{\mathrm{emb}}_{k}}}^{\prime}\right\|^{2}\right],$

where the third last step and the last step come from the fact that for any
$a_{1},a_{2},\ldots a_{n}$, we have the following inequality:
$\left\|a_{1}+a_{2}+\cdots+a_{n}\right\|^{2}\leqslant
n\sum_{i=1}^{n}\left\|a_{i}\right\|^{2}$. We then bound the right side of (13)
with $\omega_{t}$ itself:

$\displaystyle\omega_{t}\leqslant$
$\displaystyle\gamma^{2}L^{2}_{\tau}\tau\sigma^{2}+\gamma^{2}\tau
L^{2}_{\tau}\mathbb{E}\left[\sum^{t-1}_{k=\mathrm{D}(t)}\left\|{{F^{\mathrm{emb}}_{k}}}^{\prime}\right\|^{2}\right]$
$\displaystyle=$
$\displaystyle\gamma^{2}L^{2}_{\tau}\tau\sigma^{2}+\gamma^{2}\tau
L^{2}_{\tau}\mathbb{E}\left[\sum^{t-1}_{k=\mathrm{D}(t)}\left\|{{F^{\mathrm{emb}}_{k}}}^{\prime}-\nabla_{\mathbf{w}^{\mathrm{emb}}}{f}\left(\mathbf{w}^{\mathrm{emb}}_{\mathrm{k}},\mathbf{w}^{\mathrm{nn}}_{\mathrm{k}}\right)+\nabla_{\mathbf{w}^{\mathrm{emb}}}{f}\left(\mathbf{w}^{\mathrm{emb}}_{\mathrm{k}},\mathbf{w}^{\mathrm{nn}}_{\mathrm{k}}\right)\right\|^{2}\right]$
$\displaystyle\leqslant$
$\displaystyle\gamma^{2}L^{2}_{\tau}\tau\sigma^{2}+2\gamma^{2}\tau
L^{2}_{\tau}\mathbb{E}\left[\sum^{t-1}_{k=\mathrm{D}(t)}\left\|{{F^{\mathrm{emb}}_{k}}}^{\prime}-\nabla_{\mathbf{w}^{\mathrm{emb}}}f\left(\mathbf{w}^{\mathrm{emb}}_{\mathrm{k}},\mathbf{w}^{\mathrm{nn}}_{\mathrm{k}}\right)\right\|^{2}+\sum^{t-1}_{k=\mathrm{D}(t)}\left\|\nabla_{\mathbf{w}^{\mathrm{emb}}}f\left(\mathbf{w}^{\mathrm{emb}}_{\mathrm{k}},\mathbf{w}^{\mathrm{nn}}_{\mathrm{k}}\right)\right\|^{2}\right]$
$\displaystyle\leqslant$
$\displaystyle\gamma^{2}L^{2}_{\tau}\tau\sigma^{2}+2\gamma^{2}\tau
L^{2}_{\tau}\mathbb{E}\left[\sum^{t-1}_{k=\mathrm{D}(t)}\omega_{k}+\sum^{t-1}_{k=\mathrm{D}(t)}\left\|f^{\prime}\left(\mathbf{w}_{k}\right)\right\|^{2}\right]$
(14) $\displaystyle\leqslant$
$\displaystyle\gamma^{2}L^{2}_{\tau}\tau\sigma^{2}+2\gamma^{2}\tau
L^{2}_{\tau}\mathbb{E}\left[\sum^{t-1}_{k=t-\tau}\omega_{k}+\sum^{t-1}_{k=t-\tau}\left\|f^{\prime}\left(\mathbf{w}_{k}\right)\right\|^{2}\right].$

Summing (14) from $t=0$ to $t=T-1$ we obtain

$\displaystyle\sum_{t=0}^{T-1}\omega_{t}\leqslant$
$\displaystyle\gamma^{2}L^{2}_{\tau}\tau\sigma^{2}T+2\gamma^{2}\tau^{2}L^{2}_{\tau}\mathbb{E}\left[\sum^{T-1}_{t=0}\omega_{k}+\sum^{T-1}_{t=0}\left\|f^{\prime}\left(\mathbf{w}_{t}\right)\right\|^{2}\right].$

Rearranging the terms, we obtain

$\displaystyle\left(1-2\gamma^{2}\tau^{2}L^{2}_{\text{emb}}\right)\sum_{t=0}^{T-1}\omega_{t}\leqslant$
$\displaystyle\gamma^{2}L^{2}_{\text{emb}}\tau\sigma^{2}T+2\gamma^{2}\tau^{2}L^{2}_{\text{emb}}\mathbb{E}\left[\sum^{T-1}_{t=0}\left\|f^{\prime}\left(\mathbf{w}_{t}\right)\right\|^{2}\right].$

By choosing a sufficiently small learning rate $\gamma$ satisfying
$1/2\geqslant\gamma\tau L_{\text{emb}}$, it follows that

(15) $\displaystyle\sum_{t=0}^{T-1}\omega_{t}\leqslant$ $\displaystyle
2\gamma^{2}L^{2}_{\text{emb}}\tau\sigma^{2}T+4\gamma^{2}\tau^{2}L^{2}_{\text{emb}}\mathbb{E}\left[\sum^{T-1}_{t=0}\left\|f^{\prime}\left(\mathbf{w}_{t}\right)\right\|^{2}\right].$

Now we get back to (12). Summing (12) from $t=0$ to $t=T-1$ we obtain

$\displaystyle\mathbb{E}\left[f\left(\mathbf{w}^{\mathrm{emb}}_{\mathrm{t}},\mathbf{w}^{\mathrm{nn}}_{\mathrm{t}}\right)-f\left(\mathbf{w}^{\mathrm{emb}}_{\mathrm{0}},\mathbf{w}^{\mathrm{nn}}_{\mathrm{0}}\right)\right]$
$\displaystyle\leqslant$
$\displaystyle-\frac{\gamma}{2}\sum_{t=0}^{T-1}\mathbb{E}\left[\left\|f^{\prime}\left(\mathbf{w}^{\mathrm{emb}}_{\mathrm{t}},\mathbf{w}^{\mathrm{nn}}_{\mathrm{t}}\right)\right\|^{2}\right]+\frac{L\gamma^{2}}{2}\sigma^{2}T+\frac{\gamma}{2}\sum_{t=0}^{T-1}\omega_{t}$
$\displaystyle\overset{\eqref{eq:organic-snapper}}{\leqslant}$
$\displaystyle-\frac{\gamma}{2}\sum_{t=0}^{T-1}\mathbb{E}\left[\left\|f^{\prime}\left(\mathbf{w}^{\mathrm{emb}}_{\mathrm{t}},\mathbf{w}^{\mathrm{nn}}_{\mathrm{t}}\right)\right\|^{2}\right]+\frac{L\gamma^{2}}{2}\sigma^{2}T+\gamma^{3}L^{2}_{\text{emb}}\tau\sigma^{2}T+2\gamma^{3}\tau^{2}L^{2}_{\text{emb}}\mathbb{E}\left[\sum^{T-1}_{t=0}\left\|f^{\prime}\left(\mathbf{w}^{\mathrm{emb}}_{\mathrm{t}},\mathbf{w}^{\mathrm{nn}}_{\mathrm{t}}\right)\right\|^{2}\right]$
$\displaystyle\leqslant$
$\displaystyle-\frac{\gamma}{4}\sum_{t=0}^{T-1}\mathbb{E}\left[\left\|f^{\prime}\left(\mathbf{w}^{\mathrm{emb}}_{\mathrm{t}},\mathbf{w}^{\mathrm{nn}}_{\mathrm{t}}\right)\right\|^{2}\right]+\frac{L\gamma^{2}}{2}\sigma^{2}T+\gamma^{3}L^{2}_{\text{emb}}\tau\sigma^{2}T,$

where the last step comes from choosing a sufficiently small learning rate
$\gamma$ satisfying $1/4\geqslant\gamma\tau L_{\text{emb}}$. Rearranging the
terms we obtain:

$\displaystyle\frac{\sum_{t=0}^{T-1}\mathbb{E}\left[\left\|f^{\prime}\left(\mathbf{w}^{\mathrm{emb}}_{\mathrm{t}},\mathbf{w}^{\mathrm{nn}}_{\mathrm{t}}\right)\right\|^{2}\right]}{T}$
$\displaystyle\leqslant$
$\displaystyle\frac{4\mathbb{E}\left[f\left(\mathbf{w}^{\mathrm{emb}}_{\mathrm{0}},\mathbf{w}^{\mathrm{nn}}_{\mathrm{0}}\right)-f\left(\mathbf{w}^{\mathrm{emb}}_{\mathrm{t}},\mathbf{w}^{\mathrm{nn}}_{\mathrm{t}}\right)\right]}{\gamma
T}+2L\gamma\sigma^{2}+4\gamma^{2}L^{2}_{\text{emb}}\tau\sigma^{2}$
$\displaystyle\leqslant$
$\displaystyle\frac{4\mathbb{E}\left[f\left(\mathbf{w}^{\mathrm{emb}}_{\mathrm{0}},\mathbf{w}^{\mathrm{nn}}_{\mathrm{0}}\right)-f\left(\mathbf{w}^{\mathrm{emb}}_{\mathrm{t}},\mathbf{w}^{\mathrm{nn}}_{\mathrm{t}}\right)\right]}{\gamma
T}+2L\gamma\sigma^{2}+\gamma L_{\text{emb}}\sigma^{2}$
$\displaystyle\leqslant$
$\displaystyle\frac{4\mathbb{E}\left[f\left(\mathbf{w}^{\mathrm{emb}}_{\mathrm{0}},\mathbf{w}^{\mathrm{nn}}_{\mathrm{0}}\right)-f^{*}\right]}{\gamma
T}+3L\gamma\sigma^{2},$

where the last step comes from the definition of $L_{\text{emb}}$ and the
second last step still comes from the our choosing a sufficiently small
learning rate $\gamma$ satisfying $1/4\geqslant\gamma\tau L_{\text{emb}}$.

By choosing the learning rate $\gamma$ to be the same as the optimal learning
rate for SGD $\frac{1}{L+\sqrt{TL}\sigma+4\tau L_{\text{emb}}}$, we obtain

$\displaystyle\frac{\sum_{t=0}^{T-1}\mathbb{E}\left[\left\|f^{\prime}\left(\mathbf{w}^{\mathrm{emb}}_{\mathrm{t}},\mathbf{w}^{\mathrm{nn}}_{\mathrm{t}}\right)\right\|^{2}\right]}{T}$
$\displaystyle\lesssim$
$\displaystyle\frac{\sqrt{L}\sigma}{\sqrt{T}}+\frac{L}{T}+\frac{\tau
L_{\text{emb}}}{T}$ $\displaystyle\leqslant$
$\displaystyle\frac{\sqrt{L}\sigma}{\sqrt{T}}+\frac{L}{T}+\frac{\tau\min\left\\{1,\alpha\right\\}}{T}$

completing the proof. ∎