Reshape: Adaptive Result-aware Skew Handling for Exploratory Analysis on Big Data

There have been extensive studies about skew handling in two major execution paradigms in big data engines – batch execution and pipelined execution. Batch-execution systems such as MapReduce (Apache Hadoop MapReduce, [n.d.]) and Spark (APACHE Spark, [n.d.]) materialize complete input data before partitioning it across workers. Pipelined-execution systems such as Flink (APACHE Flink, [n.d.]), Storm (ApacheStorm, [n.d.]) and Amber (Kumar et al., 2020) send input tuples to a receiving worker immediately after they are available. The complete input is not known to the operator in pipelined execution, which makes skew handling more challenging.

During the execution of an operator, the controller periodically collects workload metrics from the workers of the operator to detect skew (Figure 2(a)). There are different metrics that can represent the workload on a worker such as CPU usage, memory usage and unprocessed data queue size (Bindschaedler et al., 2018; Fernandez et al., 2013). Skew handling in Reshape is independent of the choice of workload metric, and we choose unprocessed queue size as a metric in this paper. We choose this metric because the results seen by the analyst depend on the future results produced by a worker, which in turn depend on the content of its unprocessed data queue. We refer to a computationally overburdened worker as a skewed worker and workers that share the load as its helper workers.

Suppose the skewed worker $S$ and its helper $H$ have been handling input partitions $I_{S}$ and $I_{H}$, respectively (Figure 2(a)). Reshape transfers a fraction of the future input of $S$ to $H$ to reduce the load on $S$. Here future input refers to the data input that is supposed to be received by a worker but has not yet been sent by the previous operator. The controller notifies $S$ about the part p of partition $I_{S}$ that will be shared with $H$ to reduce the load of $S$ (Figure 2(b)). The downstream results shown to the user have a role to play in deciding p, which will be discussed in Section 3. Worker $S$ sends to $H$ its state information $State_{p}$ corresponding to the partition p (Figure 2(c)). Details about state-migration strategies are in Section 5. We assume the state migration time to be small till Section 5. In Section 6, we consider the general case where the state-migration time can be significant. Worker $H$ saves the state information and sends an ack message to $S$, which then notifies the controller (Figure 2(d)). The controller changes the partitioning logic at the previous operator (Figure 2(e,f)).

There are broadly two approaches to transfer the load from a skewed worker to its helper worker. We use the probe input of the HashJoin operator in Figure 1 (from the Filter operator) as an example to explain the concepts in this section. It is assumed that the build phase of the join has finished. Suppose Reshape detects $J_{6}$ and $J_{5}$ as the skewed workers in the running example and $J_{4}$ and $J_{2}$ are their corresponding helpers, respectively. The load-transfer approaches are implemented by changing the partitioning logic at the Filter operator and affects the future tuples going into the HashJoin operator.

1. Split by keys (SBK). In this approach, the keys in the partition of the skewed worker are split into two disjoint sets, say $p_{1}$ and $p_{2}$. The future tuples belonging to $p_{2}$ are redirected to the helper worker, while tuples belonging to $p_{1}$ continue to be sent to the skewed worker. For example, the partition of the skewed worker $J_{6}$ is divided into $p_{1}$ = {December} and $p_{2}$ = {June}, and the future June tuples are sent to $J_{4}$, while December tuples continue to go to $J_{6}$.

2. Split by records (SBR). In this approach, the records of the keys in the partition of the skewed worker are split between the skewed and the helper worker. The ratio of the split decides the amount of load transferred to the helper worker. For example, if the Filter operator needs to redirect $\frac{9}{26}$ of the input $J_{6}$ to $J_{4}$, then it redirects $9$ tuples out of every $26$ tuples in $J_{6}$’s partition to $J_{4}$.

When SBK is used to mitigate the skew, the processing of June tuples is transferred to $J_{4}$ (Figure 3(b)). However, this transfer has little effect on the results shown to the user. The production rates of October and December after the transfer are $\frac{6}{8}*t$ and $t$ respectively. That is, the heights of the December and October bars are still about the same, which is not representative of the final results.

SBR has more flexibility for transferring load than SBK because SBR can split the tuples of a key over multiple workers. It leads to more representative initial results than SBK as shown next. The processing of December and June tuples can be split between $J_{6}$ and $J_{4}$. For simplicity of calculation, we assume that only December tuples are shared with $J_{4}$. Since December tuples are now processed by two workers instead of one, the speed of production of these tuples increases. In order to make the future workloads of $J_{4}$ and $J_{6}$ similar, SBR redirects $\frac{9}{26}$ of the input of $J_{6}$ to $J_{4}$, which increases the total percentage load on $J_{4}$ to $16$ and decreases that on $J_{6}$ to $17$. This is implemented by redirecting $9$ December tuples out of every $26$ tuples in $J_{6}$’s partition to $J_{4}$. The production rates of October tuples after the transfer is $\frac{6}{16}*t$. The December tuples are produced by $J_{4}$ and $J_{6}$. The production rate of December by $J_{4}$ is $\frac{9}{16}*t$ and by $J_{6}$ is $\frac{16}{17}*t$, which results in a total of approximately $\frac{24}{16}*t$. Thus, using SBR leads to a more representative production ratio of December to October tuples of about $24{\,:\,}6$, which is similar to the actual ratio of $25{\,:\,}6$.

SBK assures that the December key is processed by only one worker at a time. Thus, it preserves the order of December tuples in the output sent to the visualization operator (Figure 4(a)). When SBR is used, the December tuples are split between $J_{4}$ and $J_{2}$. In the example shown in Figure 4(b), the Filter operator starts partitioning December tuples by SBR when the tuples around the $20^{th}$ of December are being produced by the Filter operator. Consequently, $J_{2}$ starts receiving the tuples from the date of the $20^{th}$ December and above. As $J_{2}$ and $J_{4}$ concurrently process data, the visualization operator receives the tuples out of order, resulting in broken line chart plots as shown in the figure.

In conclusion, SBR allows more flexibility and enables the production of representative initial results than SBK, but SBR does not preserve the order of tuples. Thus, SBR can be chosen unless there exists a downstream operator that imposes some requirement over the input order of the tuples. Such operators can be found at the workflow compilation stage. The operators before such an operator in the workflow can adopt SBK.

The goal of skew mitigation is to use one of the two approaches to transfer the load from the skewed worker to the helper worker in such a way that both workers have a similar workload for the rest of the execution. The skew handling works in literature usually have a single phase of load transfer that focuses on splitting the incoming input such that the workers receive similar load in future. Reshape has an extra phase of load transfer at the beginning that removes the existing load imbalance between the workers. We first give an overview of the two phases in Reshape, and explain the significance of the first phase.

To simplify the discussion of the second phase, we make the following assumptions:

• •

  The two workers receive data at constant rates.

• •

  We have a perfect estimator to accurately predict the incoming data workload on the workers.

In Section 4 we will relax these two assumptions. In Figure 1(c), the original load ratio of $J_{6}$ to $J_{4}$ is $26{\,:\,}7$. SBK cannot handle the skew between $J_{6}$ and $J_{4}$. The approach transfers the June month to $J_{4}$ (Figure 5(c1)), which does not mitigate the skew. However, SBR can redirect $\frac{9}{26}$ of the input of $J_{6}$ to $J_{4}$, which mitigates the skew by increasing the percentage load on $J_{4}$ to $16$ and decreasing the percentage load on $J_{6}$ to $17$. An example where SBK can mitigate the skew is the case of skew between the skewed worker $J_{5}$ and its helper $J_{2}$. SBK can transfer the processing of May to $J_{2}$, which brings the two workers to a similar workload. Specifically, the percentage load on $J_{2}$ increases to $18$ and that on $J_{5}$ decreases to $15$.

It should be noted that two phases do not mean that the state transfer has to be done twice necessarily. There are implementations where the state transfer during the first phase is enough and the second phase does not require another state transfer. For example, in SBR, the state of all keys are sent to $J_{4}$ in the first phase, and there is no state migration needed for the second phase.

We measure the load reduction ($LR$) from mitigation as the difference in the maximum input size received by a skewed worker and its helper without and with mitigation. Formally, let $S$ and $H$ represent the skewed worker and the helper worker, respectively. The load reduction is defined as:

where $\sigma_{w}$ is the size of the total input received by a worker $w$ during the entire execution.

In Figure 7, $D$ represents the difference in the total input sizes of $S$ and $H$ in the unmitigated case. When mitigation is done, due to workload estimation errors, the second phase may not be able to redirect the precise amount of data to keep the workloads of $S$ and $H$ at a similar level. In Figure 7(a), less than $\frac{D}{2}$ tuples of $S$ are redirected to $H$. Thus, $S$ receives more total input than $H$ and the load reduction is less than $\frac{D}{2}$. Similarly, in Figure 7(b), more than $\frac{D}{2}$ tuples of $S$ are redirected to $H$. As a result, the load reduction is again less than $\frac{D}{2}$. The ideal mitigation, shown in Figure 7(c), makes the total input of the two workers equal so that they finish around the same time. In particular, $\frac{D}{2}$ tuples of $S$ are sent to $H$, which is the maximum load reduction ($LR_{max}$) that can be achieved.

In this subsection, we discuss how the load reduction is affected by the value of $\tau$ at which the mitigation starts. Assume that the operator can have only one iteration of mitigation consisting of two phases. If the second phase uses a perfect estimator and the incoming data rates are constant, as assumed in Section 3, then the maximum load reduction of $\frac{D}{2}$ can be achieved. That is:

where $LR_{1}$ and $LR_{2}$ are the load reduction resulting from the first phase and second phase, respectively.

In general, the workloads estimations have errors (Chaudhuri et al., 1998; Ramakrishnan et al., 2012; Chen et al., 2015; Yan et al., 2013; Gufler et al., 2012). These errors can cause the second phase to redirect less or more than the ideal amount of $S$ tuples (Figure 7(a,b)). In other words, the load reduction from the second phase depends on the accuracy of workload estimation. The workload estimation accuracy depends on $\tau$ as shown next. If $\tau$ increases, then it takes a longer time for the workload difference of $S$ and $H$ to reach $\tau$, resulting in a higher sample size. Suppose the estimation accuracy increases as the sample size increases. Then a higher $\tau$ means that the system makes a more accurate workload estimation. Thus, the total load reduction can be computed as the following:

where $f(\tau)$ is a function representing the error in the estimation of the future workloads. As $\tau$ increases, $f(\tau)$ decreases.

The above analysis shows that a higher $\tau$ results in a higher load reduction. However, setting $\tau$ to an arbitrarily high value means that the system waits a long time before starting the mitigation. Consequently, there may not be enough future input left to mitigate the skew completely. Thus the value of $\tau$ should be chosen properly to achieve a balance between a high estimation accuracy and waiting so long that the opportunity to mitigate skew is lost. This is a classic exploration-exploitation dilemma (Auer et al., 2002).

Figure 8(a) shows the relationship between $\tau$ and load reduction. A small $\tau$ results in a small load reduction because of a high estimation error. As $\tau$ increases, $f(\tau)$ decreases and load reduction increases. The load reduction cannot exceed $LR_{max}=\frac{D}{2}$. However, the load reduction does not remain at $\frac{D}{2}$ as $\tau$ further increases, as shown next. Suppose $S$ and $H$ are to receive $1,000$ and $200$ tuples in total, respectively. Figure 8(b) shows the time when they have received $600$ and $120$ tuples respectively. At this time, the remaining $400$ tuples of $S$ can be redirected to $H$ to achieve the maximum load reduction of $\frac{D}{2}$ ($=\frac{1000-200}{2}$). The difference in the workloads of the workers at this time is denoted by $\tau_{h}$. After $\tau_{h}$, the load reduction continues to decrease because there are not enough future tuples left. Ultimately, at $\tau=D$, the load reduction becomes $0$.

When the workloads of $S$ and $H$ diverge due to workload estimation errors, the controller may start another mitigation iteration. Section 4.3.1 discusses how multiple iterations of mitigation are performed. In the previous subsection, we saw that $\tau$ should be chosen appropriately to maintain a balance between workload estimation accuracy and a long delay in the start of mitigation. Section 4.3.2 shows how to autotune $\tau$ adaptively to make better workload estimations, rather than asking the user to supply an appropriate value of $\tau$.

In this subsection, we define two types of operator states, namely immutable state and mutable state. When an operator receives input partitioned by keys, the state information of keys is stored in the operator as keyed states (Carbone et al., 2017). Each keyed state is a mapping of type

An input tuple uses the state associated with the $scope$ of the key of the tuple. If the $val$ of this $scope$ cannot change, we say the state is immutable; otherwise, it is called mutable. For example, the processing of a probe tuple in HashJoin does not modify the list of build tuples for its key. Such operators whose states are immutable are called immutable-state operators. On the other hand, an input tuple to sort is added to the sorted list associated with its $scope$ (range of keys), thus it modifies the state. Such operators that have a mutable state are called mutable-state operators.

Notice that the execution of an operator can have more than one phase. For instance, a HashJoin operator has two phases, namely the build phase and the probe phase. The concept of mutability is with respect to a specific phase of the operator. In HashJoin, the states in the build phase are mutable, while the states in the probe phase are immutable. Reshape is applicable to a specific phase, and its state migration depends on the mutability of the phase. Table 1 shows a few examples of immutable-state and mutable-state operators.

Figure 10 shows how to handle state migration for operators when using the two load-transfer approaches discussed in Section 3.

The state-migration process for immutable-state operators, as shown in branch (a) in Figure 10, involves replicating the skewed worker’s states at the helper, followed by a change in the partitioning logic. Thus, the tuples redirected from the skewed worker to the helper can use the state of their $scope$ at the latter. In contrast, the state-migration process is more challenging for mutable-state operators (branch (b)) because it is difficult to synchronize the state transfer and change of partitioning logic for a mutable state (Mai et al., 2018). State-migration strategies that focus on such synchronization exist in the literature and will be briefly discussed in Section 5.3. As we show in Section 5.4, such a synchronization is not always possible. Next, we discuss how to do state migration when using the two load-transfer approaches in mutable-state operators.

The SBK approach offloads the processing of certain keys in the skewed worker partition to the helper. Consider a group-by operator that receives covid related tweets and aggregates the count of tweets per month. The skewed worker offloads the processing of a month (say, June) to the helper. There needs to be a synchronization between state transfer and change of partitioning logic so that the redirected June tuples arriving at the helper use the state formed from all June tuples received till then. In the case of group-by, this state is the count of all June tuples received by the operator. Existing work on state-migration strategies focuses on this synchronization. A simple way to do this synchronization is to pause the execution, migrate the state, and then resume the execution (Armbrust et al., 2018; Carbone et al., 2017; Shah et al., 2003). A drawback of this approach is that pausing multiple times for each iteration may be a significant overhead. Another strategy is to use markers (Elseidy et al., 2014). The workers of the previous operator emit markers when they change the partitioning logic. When the markers from all the previous workers are received by the skewed and helper workers, the state can be safely migrated. Thus, skew handling in mutable-state operators using the “split by keys” approach can be safely done by using one of these state-migration strategies (branch (b1) in Figure 10).

In this subsection, we use the SBR approach in mutable-state operators (branch (b2) in Figure 10). We show that the synchronization between state transfer and change of partitioning logic is not possible when using this approach and discuss its effects. Consider a sort operator with three workers, namely $S_{1}$, $S_{2}$, and $S_{3}$, which receive range-partitioned inputs. The ranges assigned to the three workers are $[0,10]$, $[11,20]$, and $[21,\infty]$. As shown in Figure 11(a), $S_{1}$ is skewed and $S_{3}$ is its helper. The controller asks the previous operator to change its partitioning logic and send the tuples in $[0,10]$ to both $S_{1}$ and $S_{3}$ (Figures 11(b,c)). The synchronization of state migration and change of partitioning logic by the aforementioned state-migration strategies relies on an assumption that, at any given time, the partitioning logic sends tuples of a particular $scope$ to a single worker only. When the tuples of $[0,10]$ are sent to both $S_{1}$ and $S_{3}$, this assumption is no longer valid. Worker $S_{3}$ saves the tuples from the range $[0,10]$ in a separate sorted list (Figure 11(d)). Such a scenario where the $val$ of a $scope$ is split between workers is referred to as a scattered state.

This scattered state needs to be merged before outputting the results to the next operator. Now we explain a way to resolve the scattered state problem. When a worker of the previous operator finishes sending all its data, it notifies the sort workers by sending an END marker (Figure 11(d)). When $S_{3}$ receives END markers from all the previous workers, it transfers its tuples in the range $[0,10]$ to the correct destination of those tuples, i.e., $S_{1}$ (Figure 11(e,f)), thus merging the scattered states for the $[0,10]$ range.

We specify sufficient conditions for a mutable-state operator to be able to resolve the scattered state issue. The above approach of merging the scattered parts is suited for blocking operators such as group-by and sort, which produce output only after processing all the input data. Thus, the above approach can be used by mutable-state operators if they can 1) combine the scattered parts of the state to create the final state, and 2) block outputting the results till the scattered parts of the state have been combined.

The state-migration time is assumed to be small till now. In this subsection, we study the case where this time could be significant.

Formally, suppose $t$ is the number of tuples processed by the operator per unit time, $M$ is the estimated state-migration time, and $\hat{f}_{S}$ and $\hat{f}_{H}$ are the predicted workload percentages of $S$ and $H$, respectively. The estimated difference in the tuples received by $S$ and $H$ during the state migration is $(\hat{f}_{S}-\hat{f}_{H})*t*M$. Therefore, given $\tau_{n}$, the value of $\tau_{n}^{\prime}$ can be calculated as follows:

Till now we have assumed a single helper per skewer worker. Next, we extend Reshape to the case of multiple helpers.

The input has been assumed to be bounded till now. Next, we discuss a few considerations when the input is unbounded.

We constructed workflows of varying complexities as shown in Figure 14. Workflow $W_{1}$ analyzed tweets by joining them with a table of the top slang words from the location of the tweet. This workflow is used for social media analysis to find how often people use local slang in their tweets. The tweets were filtered on certain keywords to get tweets of a particular category. Workflow $W_{2}$ was constructed based on TPC-DS query $18$, and it calculated the total count per item category for the web sales in the year $2001$ by customers whose $birth\_month>=6$. Workflow $W_{3}$ read the Orders table from the TPC-H dataset and filtered it on the orderstatus attribute before sorting the tuples on the totalprice attribute. Workflow $W_{4}$ joined the two synthetic tables on the key attribute. Figure 15 shows the distribution of the datasets that may cause skew in the workflows. Figure 15(a) shows the frequency of tweets, used in $W_{1}$, based on the location attribute. Figure 15(b) shows the distribution of the Orders table on its totalprice attribute, used in $W_{3}$, for a $100$GB TPC-H dataset. Figure 15(c) shows the distribution of the larger synthetic table in $W_{4}$ on the key attribute. Figures 15(d)-15(f) show the distribution of the three attributes of the sales table in $W_{2}$ used in the three join operations for a $1$GB dataset.

All experiments were conducted on the Google Cloud Platform (GCP). The data was stored in an HDFS file system on a GCP dataproc storage cluster of 6 e2-highmem-4 machines, each with 4 vCPU’s, 32 GB memory, and a 500GB HDD. The workflow execution was on a separate processing cluster of e2-highmem-4 machines with a 100GB HDD, running the Ubuntu 18.04.5 LTS operating system. In all the experiments, one machine was used to only run the controller. We only report the number of data-processing machines. The number of workers per operator was equal to the total number of cores in the data-processing machines and the workers were equally distributed among the machines.

We evaluated the effect of skew and the different mitigation strategies on the results shown to the user. We ran the experiment on $48$ cores ($12$ machines). California (location $6$) produced the highest number of tweets ($26$M) in the tweet dataset. Arizona (location $4$) and Illinois (location $17$) produced $3.8$M and $6.5$M tweets, respectively. In the unmitigated case, the tuples of California (CA), Arizona (AZ), and Illinois (IL) were processed by workers $6$, $4$, and $17$, respectively. We performed two sets of experiments, in which we mitigated the load on worker $6$ processing CA tweets by using different helper workers. In the first set of experiments, we used worker $4$ as the helper and monitored the ratio of CA to AZ tweets processed by the join operator. In the second set, we used worker $17$ as the helper and monitored the ratio of CA to IL tweets processed by the join operator. The line charts in Figure 16 and 17 show the absolute difference of the observed ratio from the actual ratio as execution progressed. In the tweet dataset, the actual ratio of CA to AZ tweets was $6.85$ and CA to IL tweets was $4.05$.

No mitigation: When there was no mitigation, the CA, AZ, and IL tweets were processed at a similar rate as explained in Section 3.1. The observed ratio remained close to $1$ till worker $4$ was about to finish processing AZ tweets in Figure 16 and worker $17$ was about to finish IL tweets in Figure 17. The observed ratio started to increase (absolute difference of observed ratio with actual ratio started to decrease) after that because worker $6$ continued to process CA tweets. The actual ratio was observed near the end of execution (about $416$ seconds) in the unmitigated case.

Flux: It used the SBK load-transfer approach. It had the limitation of not being able to split the processing of a single key over multiple workers. The skewed worker $6$, apart from CA, was also processing the tweets from West Virginia. The processing of the tweets from West Virginia (about $600$K) was moved to the helper worker by Flux. However, this did affect the observed ratio of tweets much.

Flow-Join: It used the SBR approach. The execution finished earlier because the approach mitigated the skew in worker $6$. Flow-Join had two drawbacks. First, it did not perform mitigation iteratively. It changed its partitioning logic only once based on the heavy hitters detected initially. Second, it did not consider the loads on the helper and the skewed worker while deciding the portion of the skewed worker’s load to be transferred to the helper. It always transferred $50$% of the load of the skewed worker to the helper. The observed ratio of tweets started increasing once skew mitigation started. It reached the actual ratio $198$ seconds in Figure 16 and around $120$ seconds in Figure 17. Due to the aforementioned drawbacks, the observed ratio of tweets continued to increase even after reaching the actual ratio because the skewed worker continued to transfer 50% of its load to the helper. The observed ratio continued to increase till it reached about $8.3$ in Figure 16 (absolute difference = $1.5$) and $6.2$ in Figure 17 (absolute difference = $2.1$). At this point, the execution was near its end and the ratio started to decrease to the actual final ratio.

Reshape: It used the SBR approach and could split the processing of the CA key with a helper worker. Reshape had the advantage of iteratively adapting its partitioning logic and considered the current loads on the helper and the skewed worker while deciding the portion of load to be transferred in the second phase (Section 3.2). Thus, Reshape kept the workload of the skewed worker and the helper at similar levels. In Figure 16 and 17, after the observed ratio reaches the actual ratio at about $120$ seconds and $130$ seconds, respectively, Reshape kept the observed ratio near the actual ratio.

We evaluated the benefits of the first phase in Reshape as discussed in Section 3.2. We followed a similar setting as in the experiment in Section 7.2 to monitor the ratio of processed tweets. There were two mitigation strategies used in this experiment. The first one was normal Reshape, with the two phases of load transfer. In the second strategy, we disabled the first phase in Reshape and just did load transfer using the second phase. The results are plotted in Figure 18 and 19.

The first phase quickly removed the existing imbalance of load between the skewed and the helper worker when skew was detected. When the first phase was present, Reshape reached the actual ratio around $120$ and $130$ seconds in Figures 18 and 19, respectively. When the first phase was disabled, Reshape reached the actual ratio around $288$ and $310$ seconds in Figures 18 and 19, respectively. Thus, the first phase allowed Reshape to show representative results earlier. Both strategies showed more representative results than the unmitigated case.

California (location $6$) produced the highest number of tweets ($26$M) and was a heavy-hitter key in the tweet dataset. We present the results for the mitigation of the skewed worker that processed the California key.

Load balancing ratio. The load balancing ratio at a moment during the execution is calculated by obtaining the total counts of tuples allotted to the skewed worker and its helper till that moment, and dividing the smaller value by the larger value. We periodically recorded multiple load balancing ratios during an execution and calculated their average to get the average load balancing ratio for an execution. A higher ratio is better because it represents a more balanced workload between the skewed worker and its helper.

The average load balancing ratio for the skewed worker that processed the California key and its helpers is plotted in Figure 20. A higher ratio is better because it represents a more balanced workload between the skewed worker and its helper worker. We ran the experiments on three settings by varying the number of cores up to $56$ (on $14$ machines), which was the total number of distinct locations.

Flux: It had the limitation of not being able to split the processing of a single key over multiple workers. Thus, the skewed worker processed the entire California input. The skewed worker was also processing another key with only a few hundred thousand tuples, which was moved to the helper when skew was detected. Flux had a low average load balancing ratio of about $0.06$.

Flow-Join: Its main drawback was the inability to do mitigation iteratively. It changed its partitioning logic once based on the heavy-hitters detected initially. The longer it spent to detect heavy-hitters with a higher confidence, the less was the amount of future tuples left to be mitigated for finite datasets. We varied the initial duration used by Flow-Join to detect heavy-hitters from $2$ seconds to $8$ seconds. When the initial time spent was $2$ seconds, the average load balancing ratio was about $0.85$ and the final counts of tuples processed by the skewed and helper workers were approximately $14$M and $12$M, respectively. On the other hand, when the duration was $8$ seconds, the ratio was about $0.6$ and the final counts were approximately $17$M and $9$M, respectively. Flow-Join was able to reduce the execution time of $W_{1}$ on $48$ cores from $416$ seconds to $302$ seconds, when the initial detection duration was $2$ seconds.

Reshape: It split the processing of the California key with a helper worker. Reshape had the advantage of iteratively changing its partitioning logic according to input distribution using fast control messages. Thus, the skewed and helper workers ended up processing almost similar amounts of data and the average load balancing ratio was about $0.92$. The execution time was reduced by 27%. In particular, Reshape was able to reduce the execution time from $416$ seconds to $302$ seconds, by mitigating the skew in $W_{1}$ running on $48$ cores.

To evaluate the effect of the latency of control messages on skew handling by Reshape, we purposely added a delay between the time a worker receives a control message and the time it processes the message. Figure 21 shows the result of varying the simulated delay from $0$ second (i.e., the message is processed immediately) to $15$ seconds on the mitigation of $W_{1}$ on $48$ cores. The figure shows the average load balancing ratio for the two pairs of skewed and helper workers processing the locations of California (location $6$) and Texas (location $48$), which had the highest counts of tweets.

Impact on responsiveness of mitigation: As the control message delivery became slower, the delay between the controller sending a message and the resulting change in partitioning logic increased. Consider the example where the controller detected a workload difference of $350$ between the skewed worker and the helper worker and sent a message to start the first phase. In the case of no delay in control message delivery, the helper worker reached a similar workload as the skewed worker within $10$ seconds. In case of a delayed delivery, the workload difference continued to increase and got larger than $350$ before the first phase was started. For example, when there was a $5$-second delay, the workload difference was at $300$ after $10$ seconds of sending the message.

Impact on load balancing. The latency in control messages affected the load sharing between skewed and helper workers. In the case of no delay, the two workers had almost similar loads and the average load balancing ratio was about $0.94$ as shown in Figures 21(a) and 21(b). As the delay increased, the framework was slow to react to the skew between workers, which resulted in imbalanced load-sharing. In the case of a $15$-second delay, the average load balancing ratio reduced to about $0.45$. Thus, low-latency control messages facilitated load balancing between a skewed worker and its helper.

We evaluated the effect of the dynamic adjustment of $\tau$ on skew mitigation in $W_{1}$ by Reshape on $48$ cores. We chose different values of $\tau$ ranging from $10$ to $2,000$, and did experiments for two settings. In the first setting, $\tau$ was fixed for the entire execution. In the second setting, $\tau$ was dynamically adjusted during the execution. The mean model estimated the workload of a worker as its expected number of tuples in the next $2,000$ tuples and the preferred range of standard error (Section 4.3.2) was set to $98$ to $110$ tuples. We allowed up to three adjustments during an execution. Whenever $\tau$ had to be increased, it was increased by a fixed value of $50$. We calculated the average load balancing ratios for the workers processing the California and Texas keys and divided them by the total number of mitigation iterations during the execution. This resulted in the metric of average load balancing per iteration, shown in Figure 22. A higher value of this metric is better because it represents a more balanced workload of skewed and helper workers in fewer iterations.

Let us first consider the cases where $\tau$ was dynamically adjusted to an increased value. Setting $\tau$ to a small value of $10$ resulted in a large number of iterations, i.e, $41$, in the fixed $\tau$ setting. In the dynamic $\tau$ setting, the controller observed that the standard error at the beginning of the second phase was greater than $110$ and increased $\tau$. Consequently, the number of iterations decreased to $14$, which resulted in a substantial increase in the metric of average load balancing per iteration. For the cases of $\tau=50$ and $100$ in the fixed setting, the average load balancing per iteration increased with $\tau$ because the number of iterations decreased. The dynamic setting slightly decreased the iteration count in these cases.

Now let us consider the case where $\tau$ remained unchanged or decreased as a result of dynamic adjustment. When $\tau=500$, the standard error was in the range $[98,110]$. Thus, the dynamic adjustment did not change $\tau$. When $\tau=1000$ in the fixed setting, the mitigation started late and the workload of skewed and helper workers were not balanced. The mitigation was delayed even more for $\tau=1500$ and $1800$ in the fixed setting and the mitigation did not happen for $\tau=2000$. In the dynamic setting for the cases of $\tau=1000$, $1500$, $1800$, and $2000$, the controller observed that the standard error went below $98$ when the workload difference was about $700$. Thus, the controller reduced $\tau$ to $700$. The advantage of dynamically reducing $\tau$ was that it automatically started mitigation at an appropriate $\tau$, even if the initial $\tau$ was very high.

We evaluated the load balancing achieved by Reshape for different levels of skew. We used $W_{2}$ for this purpose. The data distributions in Figures 15(d)-15(e) show that the join on item_id was highly skewed and the join on date_id was moderately skewed. We evaluated the load balancing achieved for these two join operators. We scaled the data size from $100$GB to $200$GB. Meanwhile, we scaled the number of cores from $40$ to $80$ and did the experiments in each configuration.

Figure 23 shows the candlestick charts of the average load balancing ratios for the top five skewed workers from each of the two joins. For the highly skewed join on item_id, the skew was detected early, and there was enough time to transfer the load of the skewed workers to the helper workers. The $25^{th}$ and $75^{th}$ percentiles of the average load balancing ratios remained above $0.6$ for all the configurations. The median of the ratios was more than $0.77$. This result shows that Reshape was able to mitigate the skew and maintain comparable workloads on the skewed and helper workers when both the input and processing power were scaled up. The join on date_id had only a moderate skew, which resulted in a delayed detection of a few of its skewed workers. Due to the delayed detection, there were fewer future tuples of skewed workers to be transferred to the helpers. Thus the ratios for the join on date_id were lower than that for the join on item_id. The performance of Reshape was also shown by the reduction in the execution time. Specifically, in the case of $40$ cores, the mitigation reduced the execution time of $W_{2}$ from $267$ seconds to $243$ seconds. In the case of $80$ cores, the mitigation reduced the time from $335$ seconds to $269$ seconds.

We evaluated how load sharing was affected when the input distribution changed during the execution. We used the synthetic dataset and workflow $W_{4}$ running on $40$ cores. Both tables in the dataset had $42$ keys. The first table contained $4,200$ tuples uniformally distributed across the keys. The second table contained $80$M tuples and was produced by the source operator at runtime. We fixed worker $0$ and worker $10$ as the skewed and helper worker, respectively. We altered the load on key $0$ and $10$, which were processed by worker $0$ and $10$ respectively. Specifically, for the first $20$M tuples, $80$% was allotted to the key $0$ and the rest $20$% was uniformally distributed among the remaining keys. For the next $60$M tuples, $60$% was allotted to the key $0$, $20$% to key $10$, and the rest was uniformally distributed. Figure 24 shows the ratio of the workloads of the helper worker $10$ to the skewed worker $0$ as time progressed. We used $\tau=2,000$ to clearly show the effects of changing distributions.

Flux. The skewed worker was processing keys $0$ and $40$. Flux had the limitation of not being able to split the processing of a single key over multiple workers. Upon detecting skew, Flux can only move the key with smaller load (key $40$) to the helper. Thus, the workload ratio of helper to skewed worker remained close to $0$.

Flow-Join: We used a $2$-second initial duration to detect the overloaded keys. Flow-Join identified key $0$ as overloaded and started to transfer half of its future tuples to the helper. Thus, the workload of the helper began to rise. At $80$ seconds (point X), the input distribution changed. Since Flow-Join cannot do mitigation iteratively, half of the tuples of key $0$ continued to be sent to the helper. The helper worker started receiving $50$% ($=60\%*0.5+20\%$) and the skewed worker started receiving $30$% ($=60\%*0.5$) of the input. Thus, the load on the helper rose and became more than the skewed worker.

Reshape: It started the first phase to let the helper worker quickly catch up with the skewed worker. Thus, the load of the helper sharply increased initially. After that the second phase started and the workload ratio got closer to $1$. At $80$ seconds (point B), the input distribution changed. At point (point C), Reshape started another iteration of mitigation and adjusted the partitioning logic according to the new input distribution. As a result, the ratio of the workloads of the workers remained close to $1$.

We evaluated the metric-collection overhead of Reshape on the workflow $W_{2}$. We scaled the data size from $100$GB to $200$GB. Meanwhile, we scaled the number of cores from $40$ (on $10$ machines) to $80$ (on $20$ machines) and did the experiments in each configuration. We disabled skew mitigation and executed $W_{2}$ with and without metric collection to record the metric-collection overhead. As shown in Figure 25, the overhead was around $1$-$2$% for all the configurations.

To evaluate its generality to other operators, we implemented Reshape for the sort operator. We used the workflow $W_{3}$ for this experiment. The Orders table was range-partitioned on its totalPrice attribute. Table 2 lists the various percentile values of the average load balancing ratio for the skewed workers that received more than $3.5$M tuples in the unmitigated case (Figure 15(b)). We scaled the data size and number of cores simultaneously from $50$GB on $20$ cores to $200$GB on $80$ cores, and did the experiment in each configuration.

As the number of cores increased, the $25^{th}$ and $75^{th}$ percentiles of the average load balancing ratios remained close to $0.9$. This result shows that the skewed and helper workers had balanced workloads when both the input and processing power were scaled up. The consistent performance of Reshape was also shown by about $20$% reduction in the execution time. Specifically, in the case of $20$ cores, the time reduced from $789$ seconds to $643$ seconds. In the case of $80$ cores, the time reduced from $809$ seconds to $667$ seconds.

We evaluated the load reduction achieved when multiple helper workers are assigned to a skewed worker. The experiment was done on $W_{1}$ running on $48$ cores. The most skewed worker among the $48$ workers received about $27$M tuples in the unmitigated case. We allotted different numbers of helpers to the skewed worker and calculated the load reduction. We set the build hash-table in each worker to have $10,000$ keys, so that the state size became significant and the state-migration time was noticeable.

The results are plotted in Figure 26. When a single helper was used, the state migration happened in $17$ seconds. The skewed worker transferred about half of its total workload to the helper, resulting in a load reduction of $13$M tuples. When $2$ helpers were used, the skewed worker transferred about two thirds of its tuples to the two helpers (about $9$M each). With more helpers, the state-migration time also increased. For $8$ helpers, the state-migration time was about $26$ seconds. Thus, there were fewer future tuples left, which resulted in a small increase in the load reduction. For $16$ helpers, the state-migration time became $32$ seconds and the load reduction decreased to $19.7$M. For $24$ helpers, the state-migration time was $39$ seconds and the load reduction decreased to $19$M.

We implemented Reshape on Apache Flink and executed $W_{1}$ on $40$, $48$, and $56$ cores. A worker was classified as skewed if its busyTimeMsPerSecond metric was greater than $80$%. Figure 27 shows the average load balancing ratio for the workers processing the California and Texas tweets. The ratio was about $0.9$, which means that the skewed and helper workers had similar workloads throughout the execution. For the $48$-core case, the final counts of tuples processed by the skewed and helper workers for California were $13$M and $14$M, respectively. The final counts of tuples processed by the two workers for Texas on $48$ cores were $10$M each. The execution time decreased as a result of the mitigation. For example, for the $48$-core case, the execution time decreased from $407$ seconds to $320$ seconds.