Fast and Fine-grained Autoscaler for Streaming Jobs with Reinforcement Learning

1 Introduction

At present, massive streaming data, e.g., video stream, is generated incessantly on service-based applications, and processed by streaming jobs [Sun et al., 2019]. Each job is comprised of multiple tasks that execute specific computing operations. In order to minimize response latency and to improve the usage efficiency of computing resources, it is critical to design an autoscaler [Nguyen et al., 2020], which is responsible for allocating computing resources to jobs or finegrained tasks, namely job-level and task-level autoscaling, respectively. Previous work has demonstrated that fine-grained resource management can generally achieve performance improvement in various computing scenarios due to a more precise resource allocation, e.g., 11-x faster execution speed for web services [Qiu et al., 2020] and 35% gain on GPU utilization [Yu et al., 2018]. It is valuable and promising to design an effective task-level autoscaler for streaming jobs.

∗Corresponding author.

Designing an optimal autoscaler is known to be NPhard [Gar´ı et al., 2021]. An easy approach is to use heuristics, namely heuristic-based autoscalers [Verma et al., 2015]. They are typically based on heuristic rules meticulously tuned by experts, which is not only a time-consuming process but also subject to human cognitive bias. To remedy these drawbacks, another line of work [Mao et al., 2016] formulates the autoscaling process as Markov Decision Process (MDP), and adopt Reinforcement Learning (RL) to train the autoscalers, namely RL-based autoscalers.

Despite the superiority of RL-based autoscalers, how to autoscale streaming jobs at task-level with RL remains a challenging problem and has not been adequately studied. As shown in Fig. 1, DeepRM [Mao et al., 2016], DREAM [Ni et al., 2020] and DeepWave [Sun et al., 2020] used RL to autoscale batch jobs that process deterministic size of input data and cannot deal with streaming jobs with time-variant workloads. Although TVW-RL [Mondal et al., 2021] considered the temporal patterns of dynamic workloads, it focused on job-level autoscaling rather than task-level, which neglected the topological task dependencies within a job and thus caused poor performance as shown in the experiments. Another obstacle is the large temporal dimension issue. Typically, the streaming jobs will be running online for months or even years [Hueske and Kalavri, 2019] and produce massive records of job states. It brings heavy computation overhead to model the recorded fine-grained task states, which will harm the autoscaling efficiency of time-critical stream computing.

To address the above issues, we propose a novel taSklevel aUtoscaler for stReaming jobs with reinforcement lEarning (named SURE). First, we give an MDP formulation that can accurately describe the autoscaling process. More specifically, as a streaming job can normally be encoded as a Directed Acyclic Graph (DAG) [Hueske and Kalavri, 2019], we organize the running states of streaming jobs as SpatioTemporal Graphs (STGs), which take both the topological task dependencies and temporal workloads evolution into consideration and can better model the dynamic task states. Second, due to the large temporal dimension issue, it is inefficient to learn the entire STG. To alleviate this problem, we further design a novel Neural Variational Subgraph Sampler to extract informative subgraphs. While the subgraph sampler can be implicitly optimized during policy training, we propose an objective function based on mutual information maximization, which can explicitly enhance and guide the sampler to extract more representative subgraphs. After that, we leverage Graph Neural Network (GNN) to learn the sampled subgraphs as the representation of the entire STG, which can accelerate the model inference speed significantly. Finally, RL is applied to train the autoscaler.

To our best knowledge, we are the first to utilize RL to learn task-level autoscaler for streaming jobs. Extensive experiments conducted on real-world datasets demonstrate that our method outperforms six baselines. In addition, ablation study, parameter sensitive analysis and visualized case study are provided for better understanding of our work.

2 RelatedWork

Heuristic-based Autoscaler. HPA [Nguyen et al., 2020] periodically configured the number of replicas based on the monitored metrics, e.g., CPU utilization ratio, and the corresponding desired metric values. Graphene [Grandl et al., 2016] identified tasks that cost a long time to complete, and selected the schedule order with the minimum runing time. Voil`a [Fahs et al., 2020] scale-up or scale-down pods by identifing high network latency or overloaded replicas.

Reinforcement-learning-based Autoscaler. Although heuristic-based methods have been widely deployed in industry, they are sub-optimal by nature. To tackle this issue, DeepRM modeled the resource states as bitmap, and applied RL to schedule resources. DeepWeave employed GNN to process DAG information and rewarded solutions with faster Job Completion Time. DREAM designed an Encoder-Decoder framework with GNN and RL to schedule jobs according to their dependencies. However, the aforementioned methods can only schedule batch jobs with static workloads. TVW-RL exploited the temporal patterns of time-varying workloads and used RL to improve the metrics for operational excellence.

Spatio-temporal GraphModeling. Spatio-temporal graph neural network is becoming growingly important in modeling time evolutionary spatial data. ASTGCN [Guo et al., 2019] consisted of three independent components to model temporal properties, where each component contained the spatio-temporal convolution and attention mechanism to effectively capture the dynamic spatio-temporal correlations. CCRNN [Ye et al., 2021] constructed learnable adjacency matrices in different layers and used a layer-wise coupling mechanism to capture the multi-level spatial dependence and temporal dynamics simultaneously.

3 ProblemDefinition

To begin with, we give a formal definition of task-level autoscaling for streaming jobs. Given a streaming job j, it can be abstracted as a DAG, denoted as 〈V, E, P〉, where V is the node set and each node denotes a task in job j, and E represents the dependencies between tasks, and P denotes how many units of resources are allocated to task nodes, namely parallelism. Job j continuously receives and processes timevarying size of input data, e.g., video stream. As shown in Fig. 2, for every L minutes (named rescaling span), the autoscaler needs to allocate more resources (scale-up parallelism) for task nodes if the workloads increase, and release free resources (scale-down parallelism) for tasks with low workloads. In the meantime, a monitor records the job DAG states every minute, which can produce L state snapshots [G1,G2, . . . ,GL] during a rescaling span, where Gl is the state of job DAG at the l-th minute in rescaling span. In Gl, the i-th task node, denoted as vi,l, is associated with a feature vector si,l ∈ RD , where the elements in si,l are system metrics recorded by monitor, including average and summation of received data size, CPU usage, parallelism and average computation time of vi,l, and D is the feature dimension. There are two typical performance metrics for streaming jobs, namely latency and resource utilization ratio. Specifically, the latency is defined as the duration of a unit of streaming data completely processed by the job, and the resource utilization ratio denotes the proportion of resource usage.

In order to train the autoscaler with RL, we use a tuple 〈State,Action, Transition,Reward〉 to formulate the autoscaling process as Markov Decision Process (MDP) that can accurately describe the dynamic task states based on the above notations. This MDP tuple can be specified as follows: · State. At the t-th step, by connecting the consecutive L snapshots recorded from the (t-1)-th to t-th step, we can build a spatio-temporal graph Gt-1 as illustrated in Fig. 2, where

• each node vi,l is associated with feature vector si,l.
• Action. Action ai,t, a decimal in (0.0, 2.0], is defined as the ratio between desired parallelism pi,t and current parallelism pi,t-1 of task node vi. This is based on an empirical setting [Taherizadeh and Grobelnik, 2020] that, for each rescaling operation, the parallelism value cannot be reduced to zero or scaled up to more than twice of previous value.
• Transition. Given the autoscaling decision ai,t conditioned on Gt-1, task node vi changes its parallelism to pi,t. The streaming job then continues running for L minutes, and generates a spatio-temporal graph Gt as the new state.

• Reward. The reward rt is designed to minimize latency lt and maximize resource utilization ratio ut:

where λ ∈ [0, 1] controls the importance of each metric and can be tuned according to specific jobs.

With the above MDP formulation, the autoscaler can make decisions based on the topological task dependencies and temporal workloads dynamics implied in the spatio-temporal graphs. However, since the rescaling span L is usually large for streaming jobs [Nguyen et al., 2020], i.e., the large temporal dimension issue, the recorded job snapshots are massive, which makes the spatio-temporal graph too huge to be directly learned with conventional Spatio-Temporal GNNs [Ye et al., 2021], and brings heavy computation overhead for state modeling and decision-making. Therefore, we need to design a graph learning method, which can accelerate the decision inference speed without performance degradation.

4 Methodology

In this section, we present the learning details of our approach. First, we propose a novel Neural Variational Subgraph Sampler. By learning the sampled informative subgraphs, it can reduce the time cost of autoscaling decisionmaking. Furthermore, in order to explicitly extract more representative subgraphs, we propose an objective function deduced by mutual information maximization. Finally, we employ RL to train the autoscaler policy.

4.1 Neural Variational Subgraph Sampler

In this part, we aim to tackle the large temporal dimension issue and effectively learn the task node representations from giant Spatio-Temporal Graph (STG). Specifically, for each task node in a job DAG, we propose to sample a subgraph from STG, which maintains the most informative features of this task node in STG. By only learning the subgraph instead of the entire STG, it can greatly save the time cost of decision making. Huang and Zitnik also have proved that the information loss of operating GNN on a subgraph rather than the entire graph is bounded by an exponentially decaying term with respect to the node number of subgraph.

More specifically, motivated by the weighted video stream sampling models [Zhi et al., 2021], we assume that there is an underlying importance weights distribution of snapshots for each task node along the temporal dimension, and only a subset of snapshots is salient and valuable. Analogously, from the spatial perspective, we also assume that only a subset of spatial neighbors is most relevant with a specific node inspired by Graph Attention Network [Veliˇckovi´c et al., 2018]. Aside from lower computation cost, another benefit of introducing the weighted sampling mechanism is to reduce the noise or irrelevant features that could lead to performance degradation. Based on the above assumptions, we design a novel Neural Variational Subgraph Sampling (named NVSS) strategy, consisting of the following four steps.

• ‹ Input Tensor Construction. Our goal is to derive the temporal and spatial sampling probability distributions p(θi|Si) and p(φi|Si), where θi is the importance weights along temporal dimension of task node vi, and φli is the spatial sampling weights of vi at the l-th snapshot, and Si denotes the running states of vi. First, for node vi, we build an input tensor Si ∈ RL×(1+K)×D from the L recorded snapshots, where K = |Nc(vi)| is the number of c-hop neighbor nodes of vi. As shown in Fig. 3, the first column (in gray) of Si is the feature vectors [si,1, si,2, . . . , si,L]T of nodes {vi,1, vi,2, . . . , vi,L}, and the last K elements (in blue) at the l-th row are the feature vectors of the c-hop neighbors of vi,l (for brevity, we hide the feature dimension D in Fig. 3, and set L = 5 and K = 4, so the shape of Si is 5 × (1 + 4)).
• ›SamplingDistribution Inference. To capture the underlying correlation of spatial and temporal domains, we aim to learn the joint spatial and temporal importance weights distribution p(Ωi|Si) = p(θi, φi|Si) instead of learning p(θi|Si) and p(φi|Si) separately. In specific, we design a variational inference network to obtain p(Ωi|Si). Here, we assume p(Ωi|Si) follows Gaussian distribution N (µi, Σi), and derive the mean µi ∈ RL×(1+K) and co-variance Σi ∈ RL×(1+K) :

where σ is activation function, andW1 andW2 are learnable parameters. Then, the joint spatio-temporal sampling distribution Ωi ∈ RL×(1+K) can be drawn from N(µi,Σi). We can obtain the θi and φli fromΩi as follows:

where θi ∈ RL and φli ∈ RK are the first column and the last K elements at the l-th row fromΩi, respectively.

fi Subgraph Sampling. Next, we sample a subgraph for node vi using θi and φi. First, k1 informative snapshots can be sampled from Multinomial(θi) for node vi. We name the vi in the corresponding sampled snapshots as temporal anchor nodes, e.g., v2,1, v2,4 and v2,5 (green nodes) in Fig. 3. Second, for each temporal anchor node vi,l, k2 spatial neighbor nodes (yellow nodes) can be sampled from Multinomial(φli ) in the l-th snapshot, where the spatial neighbor nodes are from the c-hop neighbors set of vi, i.e., Nc(vi). In this way, we can obtain the sampled temporal anchor nodes, and the corresponding spatial neighbor nodes. By connecting temporal anchor nodes as depicted in Fig. 3, we can reconstruct the sampled subgraph gi for task node vi. Under this sampling procedure, the marginal likelihood of gi is:

where vls and vl are the sampled spatial neighbor and temporal anchor nodes, respectively.

Note that the sampling processes include drawing Ωi from N(µi,Σi) and sampling nodes from p(θi) and p(φi), but they are undifferentiable, making back-propagation inapplicable. To solve this issue, we adopt the Reparameterization Trick [Kingma et al., 2015] that can be specified as follows:

where x and y are sampled random variables from Gaussian and Gumbel distribution respectively, and diag(I) is a diagonal matrix with all-one diagonal elements. In this way, we can transfer the undifferentiable factor to x and y, while µi and Σi are differentiable and can be optimized during training.

flAutoscaling DecisionMaking. After obtaining the sampled subgraph gi, we utilize GraphSAGE [Hamilton et al., 2017] to learn subgraph representation as the state of task node vi, which can save considerable amount of state modeling time compared with learning the entire STG (detailed time complexity analysis can be found in Appendix B). The message passing process of GraphSAGE can be specified as:

where We3 is the learnable parameter at the e-th layer of GraphSAGE, andM(i) is the neighbor nodes set of vi. The input h1 of the first layer of GraphSAGE is the node features {su}u∈gi . Recall that subgraph gi is sampled with respect to node vi, we can regard hgi = Readout(hEu |u ∈ gi) as the representation of vi, where the Readout function is average pooling and E is the number of GraphSAGE layers.

Finally, a linear layer acts as the agent to make autoscaling decision for vi based on the state representation hgi :

where ai is the action decision of autoscaling (introduced in Section 2), andW4 is a learnable parameter.

4.2 Enhancing NVSS viaMutual Information

In the previous section, NVSS can be implicitly learned to sample informative spatio-temporal subgraphs, due to the existing of reward signal from RL. In this section, we explicitly encourage the sampled subgraphs to reveal the most representative topological information of the entire STG, and enhance NVSS to extract more representative subgraphs.

Specifically, we would like to minimize the semantic divergence between the sampled subgraph gi and the entire STG G. Since the Mutual Information (MI) [Kraskov et al., 2004] measures the mutual dependence between two variables, and a larger MI means that the two variables are more correlated, we propose an objective function from the MI perspective as follows (we omit the subscript of gi for concision):

where H denotes the entropy of the given distribution, and f denotes the GraphSAGE that extracts features for graphs.

With the likelihood distribution of subgraph g (Eq. 2), the lower bound of objective function (Eq. 4) can be derived as follows (please see Appendix A for detailed derivation):

where KL represents KL-divergence [Anzai, 2012], and CE denotes Cross Entropy loss, and ˆΩ is the joint distribution of actually sampled temporal anchor and spatial neighbor nodes.

The objective function (Eq. 6) includes three parts, i.e., KL-divergence, log-likelihood of prior distributions and log of MI, which are intuitive. The KL-divergence term attempts to minimize the distance between the estimated distribution q and the true distribution p. The second term is regularization term of prior distributions. The log of MI ensures that the sampled subgraphs contain the most representative information of the entire STG. The first and second terms in Eq. 6 are easy to compute. As for the MI term, we adopt the JensenShannon MI estimator [Nowozin et al., 2016] as follows:

where Dw is a discriminator that takes a subgraph and STG embeddings pair as input and determines whether the subgraph is sampled from the STG, and sp(z) = log(1 + ez ) is the softplus function, and g′ is negative sampled subgraph from ˆ p = p. In practice, we generate negative subgraphs by using all possible combinations of STG and subgraph embeddings across all graph instances in a mini-batch.

Indeed, there have been a few studies that used subgraph sampling and mutual information techniques simultaneously, however, our method has two major differences. First, NVSS can capture the underlying correlation of spatial and temporal domains, and be end-to-end optimized, while the samplers in previous work are heuristic-based [Sun et al., 2021] or cannot perform sampling for a specific node along the temporal dimension [Yu et al., 2020], which cannot be applied in this scenario and compared with our method. Second, we seek for a seamless integration of the subgraph sampling and mutual information rather than a simple combination of the two techniques. With the derived objective function (Eq. 6), the two parts can be jointly trained and mutually optimized. Moreover, the objective function is also intuitive and interpretable.

4.3 Training with Reinforcement Learning

Here, we use RL to optimize the autoscaler policy. The expected sum of rewards is R = Eπ[∑∞t=0 γ t rt], where π represents the policy that makes autoscaling decisions with Eq. 3, and γ ∈ (0, 1] is a discount factor that reduces future rewards relative to current reward, and rt is the reward function as defined in Section 2. The objective function that needs to be maximized for optimizing the autoscaler policy is: J(ψ) = 1N ∑Nn=1 logπψR, where ψ is the model parameters. Combined with the MI-based objective function (Eq. 6), the total loss of our model is L = -J-λ1 ∑|V | i=1 Yi, where λ1 is a hyperparameter that adjusts the weight of the MI-based loss. More learning details can be found in Appendix C.

5 Experiments

5.1 Experimental Setup

Simulation Environment. Following the streaming job execution logic described in Section 2, we implement a simulated computing system for streaming jobs and register a monitor for each job to record its state snapshots periodically. Dataset. We use Clarknet Trace 1 as workloads, which describes the number of HTTP requests to the servers recorded in 20,000 minutes. The workloads are highly varying in time, and show periodicity characteristics. As for computing jobs, since we focus on long-running streaming jobs, we only keep the jobs in Alibaba Cluster Dataset 2 that were running for more than 2,000 minutes following Mondal et al.. The details of datasets can be found in Appendix D.1.

Experiment Settings. To evaluate the performance of our method on jobs with different task numbers, we randomly sample six jobs and divide them into three sets, namely Small, Medium and Large. More specifically, the task numbers of Small-1, Small-2, Medium-1, Medium-2, Large-1 and Large2 are 6, 16, 25, 32, 40 and 46, respectively. Each job acts as an individual simulation environment and receives a subsequence of workloads that lasts for 7 × 24 hours in ClarkNet Trace in each episode. For each job, we first train an autoscaler and then use it to perform testing on this job for ten times and take the average reward as the final result.

1 ftp://ita.ee.lbl.gov/html/contrib/ClarkNet-HTTP.html

2 https://github.com/alibaba/clusterdata

Baselines. We compare our method with baselines from three categories: (1) heuristic-based autoscaler, i.e., HPA; (2) RL-based autoscaler, i.e., DeepWave, DREAM and TVW-RL; and (3) spatio-temporal GNN, i.e., ASTGCN and CCRNN. Due to space limitation, we skip their details that have been presented in related work. For fair comparisons, we use the same features for all baselines, and make slight modifications for them to fit for our scenario. Following DREAM, DeepWave and TVW-RL, we use REINFORCE [Williams, 1992] to train our autoscaler. The details of parameter settings and modifications of all compared methods are in Appendix D.2. The code and Appendix are available at https://github.com/xmzzyo/sure.

5.2 Performance Comparison

We present the standardized rewards of all baselines and our method on Small, Medium, Large job settings in Table 1.

In general, RL-based autoscalers perform better than HPA on most jobs, because their strategies can be optimized regarding specific job environments, while HPA uses heuristic rules and cannot be well adapted to different jobs and dynamic workloads. TVW-RL performs the best among RLbased methods on average, since it can model the temporal patterns of workloads, which is critical when autoscaling streaming jobs. Among all the baselines, CCRNN achieves the best performance on average. It uses a layer-wise coupling mechanism to capture the multi-level dependence of temporal and spatial domains. Our method consistently outperforms all the competitors on all jobs. It strongly supports our claim that the proposed subgraph sampling strategy can extract representative subgraphs, which imply the joint spatial and temporal correlations of the entire STG. By only learning the subgraphs, it can derive effective task representations and make good decisions. Moreover, our sampling mechanism can also alleviate the interference of redundancy information or noise, which can benefit the model performance.

5.3 Ablation Study

To examine the effects of different modules, three variants of our method are compared, including: (A) w/o NVSS denotes using an uniformly random sampler instead of NVSS; (B) w/o

• (c) Latency (lower is better).

(d) Utilization Ratio (higher is

better).

Figure 4: Sensitivity analysis by varying k1 and λ for Large-1 job.

MI denotes using only NVSS without mutual-informationbased objective function (Eq. 6); (C) SURE denotes our complete model. Table 2 indicates that the random sampling strategy does not perform well. Nevertheless, with our subgraph sampling strategy NVSS, it can extract informative subgraphs that contribute to a higher final reward. By incorporating our proposed objective function based on MI, it can lead to a great performance improvement, which proves that this objective function can guide the NVSS to extract more representative subgraphs and learn effective task representations.

5.4 Performance Tuning

One major contribution of this paper is the subgraph sampling strategy, which can reduce the time cost of model inference, while achieving superior performance. In this part, we report the inference time and reward influenced by the number of sampled snapshots (i.e., k1). In addition, by varying λ in Eq. 1, we also examine the sensitivity of metrics that compose the reward, i.e., latency and utilization ratio. Due to space limitation, we only present the results on Large-1 job (the results on the other jobs are similar and can be found in Appendix D.3), and incorporate the best baseline CCRNN from Table 1 for comparison.

Fig. 4(a) shows that our model consistently outperforms the best baseline, and the reward declines when k1 increases. A possible reason is that redundant snapshots may introduce noise or irrelevant information that lead to performance degradation for Large-1 job. Fig. 4(b) indicates that the inference time of our model is much less than CCRNN and shows a rising trend as k1 increases, since learning more snapshots would bring more computation overhead. From Fig. 4(c) and 4(d), we can see that our approach can achieve lower latency and competitive utilization efficiency on most λ settings. Note that we cannot always have improvements on both metrics, since a lower latency would generally cost more or even redundant resources. Nevertheless, the comprehensive performances of our model, i.e., the reward and inference time, are consistently better than the best baseline.

5.5 Case Study

In this part, we show qualitative cases based on Small-1 job to give intuitive impressions of our model. The DAG structure of Small-1 job can be found in Appendix D.4

Fig. 5(a) presents the heatmap of sampling distribution Ω of task node v1. From the first column of Ω (i.e., θ), we can observe that the most recent snapshots (framed in green) are more likely to be sampled along temporal dimension, which is reasonable and intuitive since they are more related to the states at the next step. Besides, the nearest (1-hop) neighbor nodes of v1, i.e., v2, v3 and v4 (framed in blue), have larger probabilities to be sampled, since they are more topological relevant to v1. In Fig. 5(b), we use t-SNE to visualize the embeddings of subgraphs sampled for task nodes and the entire STG in ten autoscaling steps. As depicted, the embeddings are grouped into four clusters. The v2, v3 and v4 are grouped together as they are 1-hop neighbors of v1, and v5 and v6 are 2-hop and 3-hop neighbors of v1. It demonstrates that our graph embeddings can effectively present the topological features of nodes, and the subgraph embeddings are centralized and closed with the embedding of the entire STG G.

6 Conclusion

In this paper, we propose a novel approach to autoscale streaming jobs at task-level with reinforcement learning. We first organize the job states as spatio-temporal graphs and give a formal MDP formulation of autoscaling process. To efficiently learn the giant spatio-temporal graphs, we design a Neural Variational Subgraph Sampler, which can greatly save the graph learning time. Furthermore, we propose an objective function based on mutual information to guide the sampler to extract more representative subgraphs. Our experiments demonstrate the superior performance and interpretability of our approach. In the future, we will apply our method to solve other classical spatio-temporal graph modeling tasks, such as traffic forecasting and pose detection, which also suffer from the large temporal dimension issue.

Acknowledgments

We thank Jianye Hao, Dong Li, Chen Chen, and the anonymous reviewers for their insightful comments on this paper. This work was supported by the National Natural Science Foundation of China under Grant No. 61872397. Zhen Xiao is the corresponding author.