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.

Immutable-state operator |  HashJoin (Probe phase), HB Set Difference (Probe phase), HB Set Intersection (Probe phase)
---|---
Mutable-state operator |  HashJoin (Build phase), HB Group-by, RB Sort, HB Set Difference (Build phase), HB Set Intersection (Build phase), HB Set Union
Table 1: Examples of physical operators based on state mutability. $HB$ means
hash-based and $RB$ means range-based.

####  13.2 Impact of mutability on state migration

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

Figure 27:  Operator state mutability and state migration.

The state-migration process for immutable-state operators, as shown in branch
(a) in Figure 27, 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 [78]. State-migration strategies that focus on such
synchronization exist in the literature and will be briefly discussed in
Section 13.3. As we show in Section 13.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.

####  13.3 Mutable-state operators: split by keys

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 [15, 29, 103]. A drawback of this approach is that
pausing multiple times for each iteration may be a significant overhead.
Another strategy is to use markers [46]. 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 27).

####  13.4 Mutable-state operators: split by records

In this subsection, we use the SBR approach in mutable-state operators (branch
(b2) in Figure 27). 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 28(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
28(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 28(d)).
Such a scenario where the $val$ of a $scope$ is split between workers is
referred to as a scattered state.

Figure 28:  Skew handling using the “split by records” approach in the sort
operator. $S_{2}$ is omitted for simplicity.

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 28(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 28(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.

###  14 Reshape in Broader Settings

Our discussion about Reshape so far is based on several assumptions in Section
10 for simplification. Next we relax these assumptions.

####  14.1 High state-migration time

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

Precondition for skew mitigation. In the discussion in Section 10, state
migration is started immediately after skew detection. If the time to migrate
state is more than the time left in the execution, the state migration is
futile. Thus, the controller checks if the estimated state-migration time is
less than the estimated time left in the execution and only then proceeds with
state migration. The state-migration time can be estimated based on factors
such as state-size and serialization cost [131, 44]. The time left in the
execution can be estimated by monitoring the input data remaining to be
processed and the processing speed [74] or by using the historical data [58].

Dynamic adaptation of $\tau$. Suppose the adapted value of $\tau$ output by
Algorithm 1 to be used in the next iteration is $\tau_{n}$. The discussion in
Section 12.3.2 assumes that the load transfer begins when the workload
difference is around $\tau_{n}$. This is possible only when the state-
migration time is small. When the time is significant, the load transfer will
start when the workload difference becomes considerably greater than
$\tau_{n}$. In order to start the load transfer at $\tau_{n}$ (as assumed by
Section 12.3.2), the skew has to be detected earlier. Thus, we adjust the skew
detection threshold to $\tau_{n}^{\prime}$, which is less than $\tau_{n}$,
such that the state migration starts when the workload difference is
$\tau_{n}^{\prime}$ and ends when the workload difference is $\tau_{n}$
(Figure 29).

Figure 29:  Adapt $\tau$ by considering the state-transfer time.

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:

$\tau_{n}^{\prime}=\tau_{n}-(\hat{f}_{S}-\hat{f}_{H})*t*M.$

####  14.2 Multiple helper workers

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

Load reduction. The load reduction definition (Section 12.1) can be extended
for $S$ and its helpers $h_{1},\ldots,h_{n}$ as follows:

$LR=\max_{w\in\\{S,h_{1},h_{2},\ldots,h_{n}\\}}(\sigma_{w})-\max_{w\in\\{S,h_{1},h_{2},\ldots,h_{n}\\}}(\sigma^{\prime}_{w}).$

In the equation, $\sigma_{w}$ and $\sigma^{\prime}_{w}$ are the sizes of the
total input received by worker $w$ during the entire execution in the
unmitigated case and mitigated case, respectively. Suppose $T$ is the total
number of tuples received by the operator and $f_{w}$ is the actual workload
percentage of a worker $w$. In the unmitigated case, $S$ receives the maximum
total input among $S$ and its helpers, which is $f_{S}*T$ tuples. In the ideal
mitigation case, $S$ and its helpers have the same workload, which is the
average of the workloads that they would have received in the unmitigated
case. As discussed in Section 12.1, the ideal mitigation results in maximum
load reduction denoted as:

$LR_{max}=\big{(}f_{S}-\dfrac{\sum_{w\in\\{S,h_{1},h_{2},\ldots,h_{n}\\}}{f_{w}}}{n+1}\big{)}*T.$

Choosing appropriate helpers. We examine the trade-off between the load
reduction and the state-migration overhead to determine an appropriate set of
helpers for $S$. Let $h_{1},\ldots,h_{c}$ be $c$ helper candidates for $S$ in
the increasing order of their workloads. From the definition above, increasing
the number of helpers results in a higher $LR_{max}$, provided the average
workload percentage reduces. However, increasing the number of helpers may
result in higher state-migration time since more data needs to be transferred.
Suppose $L$ is the number of future tuples to be processed by the operator at
the time of skew detection. The estimated number of future tuples left to be
processed by $S$ after state migration is $F=(L-M*t)*\hat{f}_{S}$. Increasing
the number of helpers may increase the state-migration time ($M$) and thus
decrease $F$, which means that there are fewer future tuples of $S$ to do load
transfer. Thus, given a set of helpers, the highest possible load reduction
after state migration is $\chi=min(LR_{max},F)$. As we add more helpers,
$\chi$ initially increases and then starts decreasing. The set of helpers
chosen right before $\chi$ starts decreasing are appropriate. Figure 30
illustrates an example. Let $W$ be the set of helper workers, which is
initially empty. After adding $h_{1}$ to $W$, we have $LR_{max}<F$, thus
$\chi=LR_{max}$. Then, we add $h_{2}$ to $W$, which decreases $F$, and
$\chi=F$. Then, we add $h_{3}$ to $W$, which decreases $F$ further and causes
$\chi$ to start decreasing. Hence, the final set of helpers for $S$ is
$\\{h_{1},h_{2}\\}$.

Figure 30:  Choosing appropriate helpers.

####  14.3 Unbounded data

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

Load reduction and impact of $\tau$. In Section 12.1, the load reduction was
calculated based on the total input received by the workers. For the unbounded
case, the load reduction can be calculated based on the input received by the
workers in a fixed period of time. The impact of $\tau$ on the load reduction
holds for unbounded case too. A small value of $\tau$ results in high errors
in workload estimation, which leads to a small load reduction. A large value
of $\tau$ that takes too long to reach is not preferred in the unbounded case
either. If a large $\tau$ delays mitigation, it can lead to back pressure,
loss of throughput, and even crashing of data-processing pipelines. The
latency of processing can increase, causing adverse effects on time-sensitive
applications such as image classification in surveillance [66].

Merging scattered states. For bounded data, the scattered states in mutable-
state operators were merged after the operator processed all the input. For
unbounded data, the scattered states can be merged when the operator has to
output results, e.g., when a watermark is received [19].

###  15 Experiments

In this section, we present an experimental evaluation of Reshape using real
and synthetic data sets on clusters.

####  15.1 Setting

Figure 31:  Workflows used in the experiments. The operators with skew are the
join operator on location in $W_{1}$, the join operators on date_id and
item_id in $W_{2}$, the sort operator in $W_{3}$ and the join operator on key
in $W_{4}$.

Datasets and workflows. We used four datasets in the experiments. The first
one included $180$M tweets in the US between 2015 and 2021 collected from
Twitter. The second dataset was generated using the DSB benchmark [43], which
is an enhanced version of TPC-DS containing more skewed attributes, to produce
record sets of different sizes ranging from $100$GB to $200$GB by varying the
scaling factor. The third dataset was generated using the TPC-H benchmark
[117] to produce record sets ranging from $50$GB to $200$GB. The fourth
dataset was generated to simulate a changing key distribution during the
execution. It included a synthetic table of $80$M tuples and another table of
$4,200$ tuples, and each table had two numerical attributes representing keys
and values.

(a) Tweet data.

(b) TPC-H data.

(c) Synthetic data.

(d) DSB sales data (date column).

(e) DSB sales data (item column).

(f) DSB sales data (customer column).

Figure 32: Partitioning-key distributions for the datasets.

We constructed workflows of varying complexities as shown in Figure 31.
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 32 shows the distribution of
the datasets that may cause skew in the workflows. Figure 32(a) shows the
frequency of tweets, used in $W_{1}$, based on the location attribute. Figure
32(b) shows the distribution of the Orders table on its totalprice attribute,
used in $W_{3}$, for a $100$GB TPC-H dataset. Figure 32(c) shows the
distribution of the larger synthetic table in $W_{4}$ on the key attribute.
Figures 32(d)-32(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.

Reshape implementation. We implemented Reshape on top of two open source
engines, namely Amber [73] and Apache Flink (release 1.13). In Amber, we used
its native API to implement the control messages used in Reshape. Unless
otherwise stated, we set both $\tau$ and $\eta$ to $100$. We used the mean
model [111] to predict the workload of workers. In Flink, we used the
busyTimeMsPerSecond metric of each task, which is the time ratio for a task to
be busy, to determine the load on a task. We leveraged the mailbox of tasks
(workers) to enable the control messages to change partitioning logic. The
control messages are sent to the mailbox of a task, and these messages are
processed with a higher priority than data messages in a different channel.
Using these control messages, we implemented the two phases of the SBR load
transfer approach on Flink as discussed in Section 11.

Baselines. For comparison purposes, we also implemented Flow-Join [100] and
Flux [103] on Amber with a few adaptations. For Flow-Join, we used a fixed
time duration at the start to find the overloaded keys. The workload on a
worker was measured by its input queue size. For Flow-Join, after skew is
detected, the tuples of the overloaded keys are shared with the helper worker
in a round-robin manner. For Flux, after skew is detected, the processing of
an appropriate set of keys is transferred from the skewed worker to its
helper. For both Reshape and the baselines, one helper worker was assigned per
skewed worker, unless otherwise stated. Also, unless otherwise stated, Flux
used a $2$ second initial duration to detect overloaded keys. To be fair, when
running Reshape, we also had an initial delay of $2$ seconds to start
gathering metrics and subsequent skew handling by Reshape.

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.

####  15.2 Effect on results shown to the user

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 33 and 34 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$.

Figure 33:  Effect of the mitigation strategies on the ratio of CA to AZ
tweets. The ideal curve is a straight line at $y=0$. Figure 34:  Effect of the
mitigation strategies on the ratio of CA to IL tweets. The ideal curve is a
straight line at $y=0$.

No mitigation: When there was no mitigation, the CA, AZ, and IL tweets were
processed at a similar rate as explained in Section 11.1. The observed ratio
remained close to $1$ till worker $4$ was about to finish processing AZ tweets
in Figure 33 and worker $17$ was about to finish IL tweets in Figure 34. 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 not 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 33 and around $120$ seconds in Figure
34. 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 33 (absolute
difference = $1.5$) and $6.2$ in Figure 34 (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 11.2). Thus, Reshape kept the workload of the skewed
worker and the helper at similar levels. In Figure 33 and 34, 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.

####  15.3 Benefits of the first phase

We evaluated the benefits of the first phase in Reshape as discussed in
Section 11.2. We followed a similar setting as in the experiment in Section
15.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 35 and 36.

Figure 35:  Effect of first phase on the ratio of CA to AZ tweets. Figure 36:
Effect of first phase on the ratio of CA to IL tweets.

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 35 and 36, respectively. When the first phase was disabled, Reshape
reached the actual ratio around $288$ and $310$ seconds in Figures 35 and 36,
respectively. Thus, the first phase allowed Reshape to show representative
results earlier. Both strategies showed more representative results than the
unmitigated case.

####  15.4 Effect of heavy-hitter keys

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 37. 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.

Figure 37:  Evaluating different methods of handling heavy-hitter keys in
$W_{1}$ using tweets. The three Flow-Join bars correspond to the initial delay
of $2$, $4$, and $8$ seconds.

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.

####  15.5 Effect of latency of control messages

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 38 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.

(a) California data.

(b) Texas data.

Figure 38: Effect of control message delay ($W_{1}$ on 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 38(a) and 38(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.

####  15.6 Benefit of dynamically adjusting $\tau$

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 12.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 39. A higher value of
this metric is better because it represents a more balanced workload of skewed
and helper workers in fewer iterations.

Figure 39:  Benefit of dynamically adjusting $\tau$ ($W_{1}$ on tweets).

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.

####  15.7 Effect of different levels of skew

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
32(d)-32(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 40:  Effect of different levels of skew ($W_{2}$ on DSB data). Each
candlestick body represents the $25^{th}$ to $75^{th}$ percentile.

Figure 40 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.

####  15.8 Effect of changes in input distribution

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 41 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.

Figure 41:  Effect of changes in input data distribution on load sharing
($W_{4}$ on the synthetic dataset).

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

####  15.9 Metric-collection overhead

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 42, the overhead was around $1$-$2$%
for all the configurations.

Figure 42:  Metric-collection overhead ($W_{2}$ on DSB data).

####  15.10 Performance of Reshape on sort

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
32(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.

# workers |  $P_{1}$ |  $P_{25}$ |  $P_{50}$ |  $P_{75}$ |  $P_{99}$
---|---|---|---|---|---
20 |  0.90 |  0.92 |  0.93 |  0.935 |  0.95
40 |  0.84 |  0.87 |  0.89 |  0.90 |  0.91
60 |  0.83 |  0.85 |  0.90 |  0.91 |  0.92
80 |  0.83 |  0.84 |  0.86 |  0.87 |  0.90
Table 2: Average load balancing ratios when Reshape is applied on sort
($W_{3}$ using the TPC-H data).

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.

####  15.11 Effect of multiple helper workers

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.

Figure 43:  Effect of multiple helper workers ($W_{1}$ on tweets).

The results are plotted in Figure 43. 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.

####  15.12 Performance of Reshape on Flink

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 44 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.

(a) California data.

(b) Texas data.

Figure 44: Mitigation by Reshape on Flink ($W_{1}$ on tweets)

###  16 Conclusions

In this chapter, we presented a framework called Reshape that adaptively
handles partitioning skew in the exploratory data analysis setting. We
presented different approaches for load transfer and analyzed their impact on
the results shown to the user. We presented an analysis about the effect of
the skew-detection threshold on mitigation and used it to adaptively adjust
the threshold. We generalized Reshape to multiple operators and broader
execution settings. We implemented Reshape on top of two big data engines and
presented the results of an experimental evaluation.

##  Chapter 4 Maestro: Result-aware Scheduling for Exploratory Data Analysis
on Big Data

###  17 Introduction

Big data processing systems have become quite popular in the last two decades.
We have mentioned various examples of such systems in the previous chapters.
We also discussed that these systems accept workflows broadly through two
interfaces - programming APIs and GUIs. The input workflow is compiled and
executed by the engine.

The task to execute an input workflow to completion is performed by a
scheduler. Scheduling a workflow involves decisions such as how to execute the
various operators in the workflow and the resources to execute them on. A
scheduler analyzes a workflow and comes up with an order to execute the
operators. It also decides how many resources (such as CPU cores and memory)
will be needed to execute these operators. Frameworks such as LSched [101] and
Decima [79] can help the scheduler decide the appropriate amount of resources
to be assigned. Once the resource needs are determined, the scheduler may need
to apply for a lease of the resources on shared clusters. Cluster managers
such as YARN [121] and Borg [122] have been developed to provide such leases.

The division of a workflow into multiple parts, each containing one or more
operators, for scheduling depends on the type of execution. In batch-based
execution adopted by systems such as Apache Hadoop [11] and Apache Spark [13],
the workflow is broken down at data-shuffle boundaries [110] into a sequence
of map and reduce phases. All the tasks of a phase should finish before the
next phase starts. There are some optimizations to allow the reduce phase to
partially overlap with the map phase using a slow-start configuration [60]. On
the other hand, in pipelined execution adopted by systems such as Apache Flink
[10] and Hyracks [27], the workflow is broken down at blocking links into a
sequence of pipelined regions [50]. Blocking links are edges in a workflow DAG
whose destination operator does not produce any output till it processes all
the input data on that edge. Examples of blocking links are the build input of
a two-phase HashJoin operator and the input to GroupBy operator.

We discussed the benefits of pipelined execution in exploratory data analysis
in Chapter 3. Pipelined execution allows the results to be produced quickly
because an operator does not wait for its entire input data to be produced
before processing the input and sending results to its downstream operators.
For example, when a workflow containing a two-phase HashJoin operator is
executed in a pipelined mode, the HashJoin operator starts executing and
producing results as soon as the upstream operator producing probe input
outputs initial results. In contrast, when the workflow is executed in a batch
mode, the HashJoin operator waits for the entire probe input to be produced
before joining the two inputs. Note that in both the execution models, the
HashJoin operator does not produce any output while processing the build
input. In this chapter, we consider the pipelined execution setting to enable
the users to see results quickly.

The problem of scheduling has been extensively studied in the literature,
mainly from the perspective of increasing the end-to-end performance [93, 7,
36]. However, there is little research on the following important problem:

> In exploratory data analytics, how to consider the timing of results shown
> to the user when scheduling a workflow?

A few works are related to this problem. A work ranks the partitions of input
data in the order of their likelihood to produce results and higher ranked
partitions are processed first [47]. Other works focus on a particular
operator such as join and customize the implementation of the operator so that
it produces results quicker [119, 75].

In this chapter, we take a different perspective. Considering a workflow DAG
as an input in a pipelined execution setting, we explore how to divide the
workflow into smaller parts that can be separately scheduled and how to
schedule the different parts so that results are shown to the user quickly.

A simple way to divide a workflow into parts for scheduling is based on the
blocking links as discussed above. However, there may exist operators that
expect the input data to arrive on their input links in a particular order and
cause exceptions if that expectation is not met. For example, a two-phase
HashJoin operator requires the complete input to arrive on its build input
link before input data arrives on its probe input link. To cater to such a
join operator, the part of the workflow producing build input should execute
completely before the part of the workflow producing probe input executes.
However, in many cases, the build and probe input are produced by the same
part of the workflow. Figure 45 shows such a workflow. Consider the execution
of this workflow in the pipelined mode. When the Scan operator starts
producing data, both Filter1 and Filter2 process data and output the results
to HashJoin simultaneously, which causes an exception. Engines typically take
a heuristic-based approach to handle such workflows. For example, Apache
AsterixDB [9] adds materialization at an output of the operator that
replicates the data, i.e., materialization is added on the link between the
Scan and the Filter2 operators. However, it can be seen in the figure that
materialization can also be added to the link between the Filter2 and the
HashJoin operator. In this chapter, we identify the different places where
materialization can be added in the workflow and examine their effect on the
results shown to the user.

Figure 45:  A workflow where probe and build inputs may arrive at join
simultaneously. This leads to an exception if the join operator expects build
input to be processed completely before the probe input arrives.

In this chapter, we present a novel result-aware scheduling framework called
Maestro that analyzes a workflow DAG to create regions in a result-aware
manner that are executed in a pipelined execution setting. We make the
following contributions. (1) Dividing the workflow into regions and creating a
region graph: We present a technique to divide a workflow into regions and an
algorithm to create a region graph that encapsulates the dependencies among
the regions of a workflow (Section 20). (2) Enumeration of materialization
choices: When there is a need for materialization, there may be multiple
places in the workflow where the materialization can take place. We present an
algorithm to enumerate all the options of materialization. (3) Result-aware
materialization choice: We perform an analysis to compare the different
choices of materialization from the perspective of result awareness.

####  17.1 Related work

Cluster resource managers. When a cluster is shared by different applications,
the applications need to interact with the resource managers to execute their
tasks. Apache Hadoop YARN [121] is a resource manager that publishes APIs that
can be used by applications to send resource requests (e.g., CPU cores,
memory) and receive resource leases to execute their tasks. Once the lease has
been obtained the application takes care of placing the tasks to be executed
on those resources. YARN allocates resources based on different policies such
as FIFO, FAIR, and capacity-based scheduling. Mesos [64] works in a fashion
similar to YARN where it offers resources to application level schedulers.
Borg [122] is a resource manager that accepts job requests and, unlike YARN
and Mesos, also manages placing the tasks of a job on different machines. In
contrast, Maestro focuses on the scheduling of a particular workflow and can
contact the cluster resource managers to ask for the computational resources
needed for execution.

Early result production. Symmetric hash join [125, 118] aims towards early
production of join results by keeping hash tables for both inputs in its
state. When a tuple arrives from one input, it is used to probe the hash table
of the other input. Early hash join [75] is a join algorithm that allows early
production of results without much overhead on the total execution time. It
runs in two phases where the first phase focuses on quickly producing results
and the second phase focuses on reducing the execution time. Another way to
produce data early is to rank the partitions of input data in the order of
their likelihood to produce results and process the higher ranked partitions
first [47]. In contrast, Maestro focuses on the execution of a workflow as a
whole and is not specific to a particular operator. The techniques in [75, 47]
can complement Maestro to get even better results.

Workflow schedulers. The schedulers in Apache Hadoop and Apache Spark divide
the workflow into a sequence of stages [110]. Each stage has a predefined
parallelism that is equal to the number of tasks in that stage. Tasks are
allocated to the computation nodes when the nodes become available [106]. The
scheduler in Apache Flink and Hyracks divides a workflow into a sequence of
pipelined regions based on blocking links [50]. The pipelined regions, similar
to stages, have predefined parallelism. In systems such as Apache Storm [14],
the traffic of data during the execution of the workflow can be monitored to
change the placement of operators and improve performance. R-Storm [93] is a
scheduler for Apache Storm that requires each task of a workflow to provide
its resource needs and allocates them accordingly. If a task receives data
from another task in the workflow, then it tries to put them on the same node.
A-scheduler [36] is a scheduler for Spark Streaming that follows a scheduling
policy that does FIFO scheduling for dependent tasks in a workflow and FAIR
scheduling for independent tasks. LSched [101] and Decima [79] are query
schedulers that use machine learning to predict how to schedule individual
queries so as to reduce the overall execution time of the workload. Davos
[104] uses a publish-subscribe pattern between operators for data transmission
and schedules operators one at a time depending on which operator has pending
input. It uses a priority-based scheduling strategy to execute the operator in
a workflow that has its input ready and is nearest to the output. Maestro is
different from the Davos scheduler because Davos assumes a publish-subscribe
based execution, whereas Maestro assumes a pipelined-based execution. Most
workflow schedulers focus on improving the performance of a workflow, i.e.,
reducing the total execution time. In contrast, Maestro approaches the
scheduling problem from the perspective of result timing awareness.

###  18 Preliminaries

In this section, we introduce a running example and a few preliminary concepts
that we refer to throughout this chapter.

Running Example. Figure 46(a) shows a workflow, which will serve as our
running example, where the user wants to analyze the effect of wildfires on
people’s perception of climate change. The user wants to analyze the text in
the tweets from before, during, and after the wildfire season. The historical
information about the count of past fires in various zipcodes is read by Scan1
operator. The zipcodes with no history of wildfires are filtered out. The
resulting historical information is joined with the tweets on the zipcode
attribute. The join is a two-phase HashJoin operator, where the historical
information is the build input and the tweet data is the probe input. The
output of the join operator is input into an ML operator that classifies the
tweet text as being related to climate change or not. The result is plotted as
stacked bar charts where the $x$ axis is the income range of zipcode and the
$y$ axis is the proportion of tweets related and unrelated to climate change.
For the tweets collected during the wildfire season, only those that have the
word “fire” are input into the join operator. The tweets during the wildfire
season are also plotted on a scatterplot operator to show their spatial
distribution.

Figure 46:  Analyzing climate change awareness in tweets before, during and
after the wildfire season. The build inputs to join are shown in red. All
joins are on the zipcode column. The ML operators determine if the tweet is
about climate change.

As discussed in Chapter 2, each operator physically corresponds to a set of
workers that process input data and send output data to the workers of the
downstream operators. There is a centralized controller that oversees the
execution of the workflow. Next we define a few preliminary terms and concepts
that will be used in this chapter.

######  Definition 4.1 (Result operators).

We use the term result operators to refer to operators that produce tabular or
visual results. These operators do not have any output links.

In Figure 46, the bar chart and the scatterplot operators are result
operators.

######  Definition 4.2 (Blocking and pipelined links).

We call a link between two operators as blocking if the destination operator
does not produce any output till it processes all the input data on that link.
We call a link as pipelined if the destination operator starts producing
output before it has processed all the input data on that link.

For example, the build input link to a two-phase HashJoin is a blocking link
because the entire build input is needed to create a complete build hash table
and no output is produced by the operator before that. The probe input link to
a two-phase HashJoin is a pipelined link because the operator may start
producing output before it has processed the entire probe input. Sort and
GroupBy operators are examples of operators that have just a single blocking
input and no pipelined inputs.

Implementation of a HashJoin operator. In this chapter, we assume the specific
implementation of the HashJoin in Amber. The HashJoin operator is implemented
as a collection of workers and each worker performs both build and probe
phases of the operator. For example, in Figure 47, $J1$ and $J2$ process the
build input from $B1$ and $B2$ in the build phase and the probe input from
$P1$ and $P2$ in the probe phase. Although this chapter uses HashJoin in most
of its examples, the results can be generalized to other operators that expect
certain inputs to be processed before other inputs. For example, the ideas in
this chapter can also be used to schedule an ML operator that takes two models
as two inputs before processing the test data that arrives on a third input.

Figure 47:  Implementation of HashJoin assumed in this chapter. The HashJoin
operator is implemented as a set of workers and each worker performs both
build and probe phases of the operator.

###  19 Overview of Maestro scheduler

A workflow DAG input into the engine needs to be scheduled for execution by
the scheduler. This section gives a high level overview of the working of the
Maestro scheduler. The Maestro scheduler is a module inside the controller of
the workflow as shown in Figure 48.

Figure 48:  Scheduling in Maestro

The user submits a workflow to the data processing engine through an
interface. The region builder compiles the workflow DAG to create a region
graph. A region is a sub-DAG of the workflow and the region graph is a graph
capturing the order in which the regions of a workflow should be executed. We
discuss the concepts of region and region graph in detail in Section 20. The
region graph is passed to the scheduler. The scheduler executes the regions
according to the configured scheduling policy.

The scheduler receives certain events from the user or the workers executing
the operators. Upon the receipt of an event, the scheduler makes scheduling
decisions based on the policy. For example, when the user sends a command to
start the workflow, the scheduler receives an event to begin the execution of
the workflow. During the execution of a workflow, the scheduler receives
various events about the execution status of the regions. An example of an
event received during execution is the completion of a worker of an operator.
The scheduler uses these events to decide on the scheduling of other regions.

###  20 Building an acyclic region graph in Maestro

In this section, we describe how an input workflow is scheduled in Maestro.
First, we define the concepts of region, dependencies between regions, and
region graph (Section 20.1). In order to schedule a region graph, it should be
acyclic. We show that in certain cases workflows may produce a cyclic region
graph that cannot be scheduled. We also discuss the solutions to modify such
workflows so that the region graph of the modified workflow is acyclic
(Section 20.2). Lastly, we discuss the algorithm that takes in a workflow DAG
and produces an acyclic region graph (Section 20.3).

####  20.1 Regions and their dependencies

Maestro divides a workflow DAG into sub-DAGs called “regions”, each of which
can be separately scheduled. We define the concepts of source operators,
regions, dependencies between regions, region graph and other related terms
below.

######  Definition 4.3 (Source operator).

Source operators are the operators in a workflow that have no pipelined input
links. In other words, the operator should have either no input links at all
or only blocking input links.

The operators with no inputs are the scan operators that produce data in the
workflow. Examples of operators with only blocking inputs are GroupBy and
Sort. We use the concept of source operators to define regions as we will see
soon.

Let us consider the following example. Figure 49 shows a workflow to create a
scatterplot of the selling price (sp) and the cost price (cp) of items
belonging to different categories and compare the profit margins for the
items. The probe input (Scan2) containing the selling prices of items is
joined with the build input (Scan1) containing the cost price of the items on
the item attribute and the output is projected to have the item, category, and
sp attributes. The output from Scan1 and Project are merged in the Merge
operator, which adds an extra column specifying if the tuple is from Scan1 or
Project. The merged data is visualized on a scatter plot.

Figure 49:  Workflow to understand regions. The blocking edges are shown in
red. The HashJoin requires the build input to be processed before the probe
input arrives.

A way to schedule the workflow without causing any exception is to first send
a start message to the Scan1 operator. The data from Scan1 goes into HashJoin
as the build input and also to the Merge operator that processes it and passes
the output to the Viz operator. After Scan1 has produced data completely and
the data has been processed by the HashJoin operator, a start message is sent
to the Scan2 operator that provides the probe input to HashJoin. The output of
HashJoin then goes to the Viz operator through the Project and Merge
operators. Thus, we define a region as the following.

######  Definition 4.4 (Region).

A region is a sub-DAG of the workflow that starts at a source operator and
contains all the operators reachable from the source operator using only
pipelined edges.

The regions for the workflow in Figure 49 are shown in Figure 50(a). Figure
50(b) shows the region graph for the workflow. We will discuss the concept of
a region graph soon. When a Scan operator in the workflow shown in the figure
receives a command from the controller to start producing tuples, its output
tuples propagate to all the operators that are reachable from the Scan
operator by pipelined links, and these reachable operators execute
simultaneously with the Scan operator.

Figure 50:  Regions in the workflow in Figure 49.

Note that the HashJoin operator in the Figure 50 is active when region $r_{1}$
is executing so that it can receive and process the output of Scan1. Systems
such as Hyracks include HashJoin in both $r_{1}$ and $r_{2}$ [27]. In this
chapter, we do not consider HashJoin to be a part of $r_{1}$ for the purpose
of simplicity of explanation. This way of representing regions is also adopted
by systems such as Flink where operators do not belong to the region of their
upstream operators producing blocking input [50].

######  Definition 4.5 (Execution of a region).

The execution of a region refers to the deployment of the operators of the
region on the cluster and the subsequent processing of input data by the
operators.

A region is executed by the controller by sending a start message to its
source operators. If a region has not been executed yet, it is called an
unscheduled region.

######  Definition 4.6 (Completion of a link and region).

A link between operators is said to be completed if all the data on the link
has been received by the destination operator. A region is said to be
completed if all its operators have processed all their input data and
produced all their output data.

######  Definition 4.7 (Dependency between regions).

A region $Y$ is said to be dependent on another region $X$ if there exists at
least one blocking link $l$ from an operator in $X$ to an operator in $Y$ and
all the data on $l$ should be completely received before the operators in $Y$
start executing.

In Figure 50(a), regions $r_{2}$ depends on $r_{1}$ because $r_{1}$ produces
the build input for the HashJoin operator in the regions $r_{2}$.

######  Definition 4.8 (Ready region).

Let a region $r$ depend on a set of regions represented by $D$. The region $r$
is said to be ready when all the regions in $D$ have completed and all
blocking links to $r$ from the regions in $D$ have also completed. When a
region is ready, it can be executed.

When a region has to be scheduled, all the regions that it depends on should
have already completed. Thus, the scheduler has to follow an order when
executing the regions of a workflow. This order is captured in the region
graph for a workflow, which we define below.

######  Definition 4.9 (Region graph).

The region graph for a workflow is a graph with the regions of the workflow as
its vertices and the dependencies between the regions as its edges. An edge
from a region $X$ to a region $Y$ in the region graph means that $Y$ is
dependent on $X$.

The region graph for the workflow in Figure 50(a) is shown in Figure 50(b). We
also show the regions in the workflow in Figure 46 and the corresponding
region graph in Figure 51.

Figure 51:  Regions and region graph from the running example workflow.
Blocking input links are in red.

####  20.2 Avoiding cycles in region graphs

The regions must be scheduled in a topological order based on the region
graph, in order to ensure that a region is executed only when all the regions
it depends on have been executed. In order to obtain a topological order, the
region graph should not have any cycles, i.e., it should be a DAG. As we show
soon, a workflow may yield a region graph with cycles. Such a workflow has to
be modified in a way that the modified workflow yields an acyclic region
graph. Next we show an example of a workflow that yields a cyclic region
graph. We also discuss ways to modify the workflow so that the modified
workflow has an acyclic region graph.

Example workflow with cyclic region graph. Let us see an example of a workflow
that yields a cyclic region graph. Figure 52 shows a workflow DAG and its
corresponding region graph. The workflow tries to find a new department for
the employees in the machine learning department and their managers. It
visualizes the count of employees and managers by the new departments they
have been allotted to. There is only one region in the workflow whose source
is the Scan operator. The HashJoin expects to first completely receive and
process its build input from Filter1. Then, it expects to receive the probe
input from Filter2. Since both Filter1 and Filter2 belong to the same region
$r_{1}$, the region graph has a cycle, as shown in the figure. Such a cyclic
region graph cannot be scheduled.

Figure 52:  Workflow with cyclic region graph. When there is a cycle in the
region graph, there is no feasible schedule of regions.

Modifying the workflow to get an acylic region graph. In order to enforce the
requirement of the probe input of HashJoin arriving after the build input, the
two inputs need to be produced by different regions. The region producing the
build input will need to be scheduled first, followed by the region producing
the probe input. There are different ways to modify the workflow so that the
probe input for HashJoin arrives from a different region than the build input.
One way is to remove the Duplicate operator and replicate the operators Scan1
and ML1 to supply the data to Filter2 as shown in Figure 53(a). This creates
two regions $r_{1}$ and $r_{2}$, and $r_{2}$ has a dependency on $r_{1}$. This
way incurs the overhead of repeated computation. For example, in Figure 53(a),
the input data is processed by the same ML operator twice.

Figure 53:  Modifying workflow in Figure 52 to create an acyclic region graph.

Another way to create a new region is to introduce materialization.
Materialization is introduced on a link by adding a materialization writer and
a materialization reader operator to the link. The materialization writer
writes all the data produced by the source operator of the link to a memory or
disk. The materialization reader reads the data from the storage and outputs
it to the destination operator of the link. In Figure 53(b), materialization
writer and reader operators have been put between the operators Filter2 and
HashJoin. The introduction of materialization also creates two regions $r_{1}$
and $r_{2}$, where $r_{2}$ is dependent on $r_{1}$. The Scan operator is the
source of $r_{1}$ and the materialization reader operator (MR) is the source
of $r_{2}$. We focus on the case where computational resources are more
expensive than storage. We will follow the materialization way to modify the
workflow DAG in case of cyclic region graphs.

####  20.3 Creating an acyclic region graph

In Section 20.2, we saw the problem of cyclic region graphs and discussed ways
to modify the workflow to yield an acyclic region graph. In this section, we
discuss an algorithm (Algorithm 2) that takes a workflow DAG $W$ as input and
yields an acyclic region graph $\mathcal{G}$. If needed, the algorithm may
modify the workflow DAG $W$ by introducing materialization operators to yield
the acyclic region graph. We use the workflow in Figure 52(a) to explain the
various steps of the algorithm.

Figure 54:  Diagram to help explain the intuition behind lines 2 to 2 in
Algorithm 2. Regions are marked by dotted blue shapes.

Input: $W\leftarrow$ A workflow DAG

Output: $\mathcal{G}$: A region graph

1 $\mathcal{S}\leftarrow$ Set of source operators $W$

2 Initialize $\mathcal{G}$ to an empty graph

3 Initialize $\mathcal{R}$ to an empty set of regions

4 $O\leftarrow$ A topological order of the operators in $W$

5 foreach  _operator o in O_ do

6 $s_{o}\leftarrow$ set of sources in $\mathcal{S}$ from which $o$ is
reachable using pipelined links

7 foreach  _source $s$ in $s_{o}$_ do

8 if  _not(there exists a region in $\mathcal{R}$ with $s$ as the source)_
then

9 Add a new region $r$ in $\mathcal{R}$ with $s$ as the source

10 Add a vertex in $\mathcal{G}$ corresponding to $r$

11

12 end if

13 Add operator $o$ to the region in $\mathcal{R}$ that has $s$ as source

14

15 end foreach

16 if  _blocking input link b exists to o_ then

17 $u\leftarrow$ Upstream operator of link $b$

18 while  _true_ do

19 $SR_{u}\leftarrow$ Set of regions that $u$ belongs to

20 $SR_{o}\leftarrow$ Set of regions that $o$ belongs to

21

22 if  _( $\exists r_{u}$ in $SR_{u}$ and $\exists r_{o}$ in $SR_{o}$ such
that adding edge $r_{u}\rightarrow r_{o}$ causes a cycle in $\mathcal{G}$)_
then

23 call Add-Materialization($W$, $\mathcal{R}$, $\mathcal{G}$, $r_{u}$,
$r_{o}$, $o$)

// This subroutine is discussed in Algorithm 3. It adds materialization to
region $r_{o}$. The materialization affects the structure of region $r_{o}$
and possible other regions too. This subroutine updates $\mathcal{R}$ and
$\mathcal{G}$.

24

25 end if

26 else

27 foreach  _$r_{u}\in SR_{u}$_ do

28 foreach  _$r_{o}\in SR_{o}$_ do

29 Add edge $r_{u}\rightarrow r_{o}$ to $\mathcal{G}$

30

31 end foreach

32

33 end foreach

34 break

35

36 end if

37

38 end while

39

40 end if

41

42 end foreach

return $\mathcal{G}$

Algorithm 2 Acyclic region graph creation.

Let $\mathcal{S}$ be the set of all the source operators in the workflow (line
2). For the workflow in Figure 52(a), there is only one source, which is the
Scan operator. The region graph $\mathcal{G}$ is initially empty (line 2). The
set of regions $\mathcal{R}$ is also initially empty (line 2). Let $O$ be a
topologically ordered sequence of operators in $W$ (line 2). A topological
order for the operators in the workflow in the figure is $\bigl{[}$Scan, ML1,
Duplicate, Filter1, Filter2, HashJoin, Viz$\bigr{]}$. The operators in $O$ are
sequentially considered and the following is done for each of them. Let the
current operator in the sequence be $o$. Let $s_{o}$ be the set of sources
from which $o$ is reachable using pipelined edges (line 2). For each source
$s$ in $s_{o}$, we check if the region corresponding to the source $s$ is
present in $\mathcal{R}$. If it is not present, we add a region $r$ that has
$s$ as the source in $\mathcal{R}$ (line 2). We also add a vertex representing
$r$ in $\mathcal{G}$ (line 2). In the example workflow, the Scan operator is
considered first. Since the region that has Scan as the source is not present
in $\mathcal{R}$ and $\mathcal{G}$, it is added to both. We denote this region
as $r_{1}$ in our example. The next operators in the order are ML1, Duplicate,
Filter1, and Filter2 and they are considered one by one. All these operators
only have one source from which they are reachable (Scan) and the region
corresponding to that source ($r_{1}$) is already in $\mathcal{R}$ and
$\mathcal{G}$.

Operators such as HashJoin have one blocking input link. On the other hand, a
machine-learning operator that accepts two models on blocking inputs before
processing test tuples using the models has two blocking input links. For
simplicity, in the algorithm, we assume that the operator has at most one
blocking input link. Let the blocking input link (if it exists) be $b$. We
denote the set of regions that the upstream operator (say, $u$) of $b$ belongs
to as $SR_{u}$ (line 2). We denote the set of regions that the operator $o$
belongs to as $SR_{o}$ (line 2). Figure 54 shows a case where $u$ belongs to
two regions ($r_{u}$ and $r_{y}$) and $o$ belongs to two regions ($r_{o}$ and
$r_{x}$). An edge needs to be added from each region in $SR_{u}$ to each
region in $SR_{o}$. If one of these edges causes a cycle in $\mathcal{G}$, a
subroutine is called to add materialization (lines 2 to 2). The details of
this subroutine are presented in the next section. This subroutine adds
materialization and modifies $\mathcal{R}$ and $\mathcal{G}$ to reflect the
newly created regions and edges due to materialization. Note that the set of
regions $SR_{u}$ and $SR_{o}$ may change after materialization, as we will
show soon using Figure 54. The Algorithm 2 again computes the sets of regions
$SR_{u}$ and $SR_{o}$ and tries to add edges between each pair of regions in
these sets. This process continues till edges have been added from each region
in $SR_{u}$ to each region in $SR_{o}$ without causing cycles.

In Figure 54(a), adding an edge from the region $r_{u}$ to region $r_{o}$
causes a cycle. Thus, we need to add materialization to $r_{o}$. A
materialization option is shown in the figure. This materialization breaks the
region $r_{o}$ and $r_{x}$ as shown in Figure 54(b) and creates a new region
$r_{n}$ with the materialization reader as the source. Thus, the set $SR_{o}$
changes from {$r_{o}$, $r_{x}$} (before materialization) to {$r_{n}$} (after
materialization). Edges $r_{u}\rightarrow r_{n}$ and $r_{y}\rightarrow r_{n}$
are added to the region graph.

In our example workflow in Figure 52(a), when the HashJoin operator is
considered, the regions of Filter1 and HashJoin are both $r_{1}$. Since adding
an edge from $r_{1}$ to $r_{1}$ will cause a self-loop, materialization is
added on the link between Filter2 and HashJoin. The added materialization
reader becomes the source of the region $r_{2}$. The HashJoin operator is now
in the region $r_{2}$. In this case, $SR_{u}=\\{r_{1}\\}$ and
$SR_{o}=\\{r_{2}\\}$. An edge is then added from $r_{1}$ to $r_{2}$. Once we
have gone through all the operators in the topological order, the graph
$\mathcal{G}$ is returned.

Note that Algorithm 2 also works for operators that only have one input link
which is blocking. Examples of such operators are GroupBy and Sort. For such
an operator (say $o$), the algorithm adds an edge from the region containing
the upstream operator of $o$ to the region containing $o$.

###  21 Choosing a materialization option

The Algorithm 2 in Section 20 may add materialization to the input workflow in
order to produce an acyclic region graph. In this section, we enumerate the
different options where the materialization operators can be placed in the
workflow (Section 21.1) and discuss how it affects result awareness (Section
21.2).

####  21.1 Enumerating the materialization choices

In Figure 53(b), the materialization operators were placed between the Filter2
and the HashJoin operators. However, another choice is to place them between
the ML1 and Filter2 operators. Both choices create two regions and result in
the same region graph as shown in Figure 53(b). In general, there can be
multiple choices where materialization operators can be placed in a workflow.
We are going to describe the process to enumerate all those choices.

We first give an intuitive explanation of how to enumerate the different
materialization choices by giving an example. Then, we give a formal approach
for enumerating the choices.

Figure 55:  Enumerating all the materialization choices. J1, J2, J3, and J4
are HashJoin operators. D1 and D2 are replication operators. M1 and M2 are ML
operators. S is scan, F is filter, and U is a union operator.

Intuition. Figure 55(a) shows the outline of a workflow with its regions
marked. The corresponding region graph containing a cycle is shown in Figure
55(b). The problem exists at the operator J4 that receives its probe input
from the region $r_{2}$ before the build input arrives from the region
$r_{4}$. In order to solve this problem, we use materialization to modify the
workflow DAG in such a way that the probe input to J4 can be delayed. The
materialization should split the region $r_{2}$ into two parts, thereby
“cutting off” the flow of data from $r_{2}$ into the probe input of J4. This
is similar to the network flow problem where one needs to determine the cut
edges in the graph that make a target vertex unreachable from the source
vertices [52]. In our case, the target vertex is the operator whose probe
input has to be cut off (J4) and the cut edges are the links where data has to
be materialized. Figure 55(c) shows materialization added to the link between
U and J4. It cuts off the probe input to J4 and divides the region $r_{2}$
into two parts $r_{2}^{1}$ and $r_{2}^{2}$ as shown in Figure 55(d).

A key consideration while adding the materialization is that the
materialization should cut off the probe input to J4 from $r_{2}$, but the
build input to J2 (in $r_{3}$) should still come from $r_{2}$. If this does
not happen and the build input to J2 is produced by the newly created region,
then there still will be a cycle in the region graph. Thus, we need to
identify the sub-DAG of the workflow where the materialization can happen such
that it cuts off the probe input to J4 from $r_{2}$ but does not impact the
build input to J2 from $r_{2}$. This sub-DAG for the workflow in Figure 55(a)
is shown in Figure 55(e). The figure also shows $6$ different sets of cuts,
each of which disconnects D1 and J4. These cuts form the $6$ materialization
choices. Each cut includes a set of edges where the data has to be
materialized. The choices $1$ and $6$ consist of one edge only, which means
that they require a single materialization writer and reader operators. When
one of these two options is chosen, the region $r_{2}$ is broken into
$r_{2}^{1}$ and $r_{2}^{2}$. The single materialization reader becomes the
source of the region $r_{2}^{2}$ and the resulting region graph is shown in
Figure 55(f). The choices $2$, $3$, $4$, and $5$ consist of two edges each,
which means that they require two materialization writer and reader operators,
one pair on each edge. When one of these four options is chosen, the region
$r_{2}$ is broken into three regions $r_{2}^{1}$, $r_{2}^{2}$ and $r_{2}^{3}$.
The two materialization readers become the source of the regions $r_{2}^{2}$
and $r_{2}^{3}$, and the resulting region graph is shown in Figure 55(f).

Formal approach. We present an algorithm to describe how materialization is
added to the workflow DAG in the line 2 of Algorithm 2. Consider that $W$,
$R$, and $\mathcal{G}$ are the workflow DAG, set of regions and the region
graph, respectively, before that line and $O$ is the operator that is
responsible for an edge from a region $r_{u}$ to $r_{o}$ that would result in
a cycle in the region graph $\mathcal{G}$. Algorithm 3 adds materialization to
$W$ to obtain a modified workflow DAG $W^{\prime}$ and prevents the creation
of a cycle in $\mathcal{G}$. Figure 56(a) shows an example outline of the
workflow $W$ with region $r_{o}$ and $r_{u}$ marked. Let $r_{m}$ be the region
after $r_{o}$ in the cycle created by $O$ in the region graph of $W$ (line 3).
The operators in the region $r_{m}$ receives blocking input from the operators
in $r_{o}$. As explained above, the materialization should cut off the input
to $O$ from $r_{o}$ but not impact the blocking input to $r_{m}$ from $r_{o}$.

Input: $W,\mathcal{R},\mathcal{G}\leftarrow$ Workflow DAG, set of regions and
region graph before line 2 in Algorithm 2

Input: $O\leftarrow$ Current operator at line 2 in Algorithm 2 due to which
Algorithm 2 needs to add an edge from a region $r_{u}$ to a region $r_{o}$,
that results in a cycle in $\mathcal{G}$

Output: $W^{\prime}$, $\mathcal{R^{\prime}}$, $\mathcal{G^{\prime}}$: Modified
workflow DAG, set of regions and region graph after adding materialization to
$W$

1

2$r_{m}\leftarrow$ The region that is after $r_{o}$ in the cycle in
$\mathcal{G}$

3 $S_{o}\leftarrow$ The source of $r_{o}$

4 $G_{o}\leftarrow$ The sub-DAG supplying the pipelined input from $S_{o}$ to
$O$

5 $G_{m}\leftarrow$ The sub-DAG supplying the blocking input from $S_{o}$ to
the operators in region $r_{m}$

6 $G_{f}\leftarrow G_{o}-G_{m}$ // The sub-DAG where materialization can be
done without affecting input to $r_{m}$

7 $P\leftarrow$ The set of operators in $G_{m}$ that supply the pipelined
input to $O$

8 $E\leftarrow$ A cut-edge set in $G_{f}$ that makes $O$ unreachable from all
operators in $P$

9 Modify $W$ by adding materialization operators to all the edges in $E$

10 $SR_{b}\leftarrow$ The set of regions that will be broken due to
materialization

11 For each region in $SR_{b}$, remove it from $\mathcal{R}$ and the
corresponding vertex from $\mathcal{G}$

12 $SS_{b}\leftarrow$ The set of sources of the regions in $SR_{b}$

13 $SS_{m}\leftarrow$ The set of materialization readers introduced

14 foreach  _source $s\in SS_{b}\cup SS_{m}$_ do

15 Add a region $r$ to $\mathcal{R}$ representing the region that has $s$ as
the source and containing all the operators reachable from $s$ using pipelined
edges

16 Add a vertex to $\mathcal{G}$ representing the region $r$

17

18 end foreach

19 foreach  _materialization reader $t$ in $SS_{m}$_ do

20 Add an edge in $\mathcal{G}$ from $r_{u}$ to the region that has $t$ as the
source

21

22 end foreach

23 $SR_{w}\leftarrow$ the new regions that have the sources of the regions in
$SR_{b}$ as their sources

24 $SR_{r}\leftarrow$ the new regions that have the materialization readers as
their sources

25 foreach  _region $r$ in $SR_{w}$_ do

26 Add edges from $r$ to the regions in $SR_{r}$ that read a materialization
output of $r$

27

28 end foreach

29 call Add-edges-to-new-regions-in-region-graph($SR_{w}$, $SR_{r}$,
$\mathcal{G}$, $W$)

30 return modified $W$, $\mathcal{R}$, and $\mathcal{G}$

31

Algorithm 3 Adding materialization to a workflow. Figure 56:  Understanding
the formal approach to find the sub-DAG where materialization is to be added.

Let $S_{o}$ be the source of the region $r_{o}$ (line 3). Let the sub-DAG that
supplies pipelined input to $O$ from $S_{o}$ be called $G_{o}$ (line 3), as
shown in the figure. Let the sub-DAG that supplies blocking input from $S_{o}$
to $r_{m}$ be called $G_{m}$ (line 3), as shown in the figure. Let
$G_{f}=G_{o}-G_{m}$. On a high level, the graph $G_{f}$ is the part of $r_{o}$
that produces pipelined input for $O$, but does not produce the blocking input
for $r_{m}$. The input to $G_{f}$ comes from a set of operators in $G_{m}$
that are represented by $P$ in the figure. Considering the sub-DAG $G_{f}$, we
need to identify all possible cut-edges sets [52] that make the operator $O$
unreachable from the operators in $P$. Each element of the set is a
materialization choice and contains a collection of edges where
materialization can be added. We choose one materialization option and apply
it to $W$ (line 3).

Let us assume that the option shown in Figure 56(a) is chosen for
materialization and use it for further discussion. Let the set of regions
affected by the materialization be $SR_{b}$. In the figure, $r_{o}$ is the
only region affected by materialization. The vertices and regions
corresponding to these regions are removed from $\mathcal{G}$ and
$\mathcal{R}$, respectively (line 3) because materialization breaks these
regions into multiple smaller regions. Then, we begin the process of adding
the new regions to $\mathcal{G}$ and $\mathcal{R}$. For the source of each
region in $SR_{b}$, we add a new region to $\mathcal{G}$ and $\mathcal{R}$,
that contains all the operators reachable from the source using pipelined
edges (lines 3-3). These are the regions that contain the materialization
writer operators and the set of these regions is denoted by $SR_{w}$. In the
figure, $SR_{w}$ only contains $r_{o}^{1}$. Similarly, for each
materialization reader introduced, we add a corresponding region to
$\mathcal{R}$ and $\mathcal{G}$. We also add an edge in $\mathcal{G}$ from
$r_{u}$ to the newly added regions with materialization readers as the sources
(lines 3-3). In the figure, two materialization readers are introduced that
create two new regions, namely $r_{n}^{1}$ and $r_{n}^{2}$. These regions now
contain the operator $O$ that receives the blocking input from $r_{u}$. Thus,
edges are added from $r_{u}$ to the two new regions in the region graph. Let
$SR_{r}$ be the set of new regions that have the materialization readers as
the sources. Next, we iterate through each region (say $r_{w}$) in $SR_{w}$
and add an edge from $r_{w}$ to the regions in $SR_{r}$ that consume the
materialized output of $r_{w}$ (lines 3-3). In the figure, this leads to an
edge from $r_{o}^{1}$ to $r_{n}^{1}$ and $r_{n}^{2}$.

When we removed the regions in $SR_{b}$ from $\mathcal{G}$, we also removed
the corresponding edges of those regions. Thus, the edges from the regions in
$SR_{w}$ and $SR_{r}$ to the other regions in $\mathcal{G}$ have to be found
and added (line 3). For example, in the figure, the region $r_{o}$ used to
produce the blocking input to the region $r_{z}$. After materialization, the
regions $r_{n}^{1}$ and $r_{n}^{2}$ produce the blocking input to $r_{z}$.
Thus, the edges $r_{n}^{1}\rightarrow r_{z}$ and $r_{n}^{2}\rightarrow r_{z}$
need to be added to $\mathcal{G}$. The computation of these edges can be done
from scratch by iterating over all the operators in the workflow processed
till now according to the topological sequence. Another way is to compute
these edges incremental by remembering the edges to the regions in $SR_{b}$
before materialization and adjusting only those edges. At the end, the
modified $W$, $\mathcal{R}$, and $\mathcal{G}$ are returned.

####  21.2 Result-aware materialization choice selection

We saw how to enumerate different materialization choices in Section 21.1. In
this section, we discuss how to consider result awareness while choosing a
materialization option. We assume that there is a single cycle in the region
graph and there is a single result operator in the workflow. We first define
first response time, which is used as a metric for result awareness in this
chapter. Then, we discuss how to choose a materialization option that gives a
good first response time.

####  21.3 First response time

In exploratory analysis, the users want to quickly observe some initial output
from the result operators. This allows the users to find issues in the
analysis early without waiting for the entire workflow to finish executing
[104]. Consider the following scenario. In the running example, when $r_{3}$
runs, the user may observe that the scatterplot operator contains many tweets
related to other types of fires different from wildfire. Since the goal of the
analysis is to determine whether wildfires lead to climate change awareness,
the user may decide to add more keywords to the Filter2 operator to make the
tweets more relevant to wildfire.

Considering a workflow with a single result operator, we define the first
response time as follows. It is the time required from the start of execution
of a workflow to the time when the result operator has produced a specific
number of tuples (say, $d$). We denote the first response time by $\tau(d)$,
where $d$ is the number of tuples that the result operators should produce.
For simplicity, we consider $d=1$ in this chapter.

Figure 57:  The outline of a workflow showing regions. The sink operator in
the last region. For first response time, we consider the time to completely
execute all regions except the region containing the sink. For the region
containing the sink, the time to produce only a single tuple is considered in
the calculation of the first response time.

First response time for a sequence of regions. Let $r_{a}\rightarrow
r_{b}\rightarrow r_{c}\rightarrow\cdots\rightarrow r_{x}$ be a sequence of
regions (e.g., a topological order of regions from an acyclic region graph),
where the result operator is present in $r_{x}$ (Figure 57). Assume each
region in the sequence is completed before the next region begins execution.
Let $t(r_{i})$ represent the time to execute a region $r_{i}$ completely and
$t(r_{i}(1))$ represent the time to produce a single tuple from the result
operator in $r_{i}$. Then, the first response time for the sequence of regions
is,

$\tau(1)=t(r_{a})+t(r_{b})+t(r_{c})+\cdots+t(r_{x}(1)).$

####  21.4 Choosing a materialization option

We now use the above description of first response time to discuss how to
choose a materialization option. The goal is to choose an option that would
reduce the first response time.

Figure 58:  The outline of modified workflows showing regions when
materialization is added to the workflow in Figure 55(a). When there are more
than one regions containing the sink, the first response time includes the
minimum time among those regions to produce a single tuple.

Explanation using example. Let us consider the example region graph with a
cycle shown in Figure 55. When materialization options $1$ or $6$ are chosen,
two new regions $r_{2}^{1}$ and $r_{2}^{2}$ are created out of $r_{2}$, and
$r_{2}^{2}$ has the result operator (Figure 58(a)). In this case, the first
response time is:

$\tau(1)=t(r_{1})+t(r_{2}^{1})+t(r_{3})+t(r_{4})+t(r_{2}^{2}(1)).$

Note that the structure of $r_{2}^{1}$ and $r_{2}^{2}$ depend on which option
is chosen. Thus, $t(r_{2}^{1})$ and $t(r_{2}^{2}(1))$ depend on which option
is chosen. The time for the execution of other regions is not affected by the
materialization choice. When materialization options $2$, $3$, $4$ or $5$ are
chosen, three new regions $r_{2}^{1}$, $r_{2}^{2}$ and $r_{2}^{3}$ are created
out of $r_{2}$, and both $r_{2}^{2}$ and $r_{2}^{3}$ have the result operators
(Figure 58(b)). In this case, the first response time is:

$\tau(1)=t(r_{1})+t(r_{2}^{1})+t(r_{3})+t(r_{4})+min(t(r_{2}^{2}(1)),t(r_{2}^{3}(1))).$

Note again that the materialization option affects the structure of
$r_{2}^{1}$, $r_{2}^{2}$ and $r_{2}^{3}$ and consequently the times
$t(r_{2}^{1})$, $t(r_{2}^{2}(1))$ and $t(r_{2}^{3}(1))$. The execution times
of other regions are unaffected.

Query time estimation can be done to estimate $\tau(1)$ resulting from a
materialization option. For a region $r$, $t(r)$ can be predicted by
estimating the total query execution time [58, 126] for the query represented
by the operator DAG in $r$. Similarly, $t(r)(1)$ can be predicted by
estimating the time to get the first few results [68]. We choose the
materialization option among the $6$ choices that has the least predicted
$\tau(1)$.

Figure 59:  Example to help understand the generalization of first response
time when a materialization choice $c$ is chosen in a region $r$. The region
$r_{c}^{1}$ has to be fully executed, and any of the regions $r_{c}^{2}$,
$r_{c}^{3}$, $\cdots$ has to produce a single tuple to get the first tuple out
of the sink.

Generalization. In general, a materialization choice $c$ breaks a region $r$
into two sets of regions: 1) a single region $r_{c}^{1}$ that has the source
of $r$ as its source and 2) a set of regions
$\\{r_{c}^{2},r_{c}^{3},\cdots\\}$ that have the materialization readers as
their sources. Let $\tau_{c}(1)$ be the first response time when the option
$c$ is chosen. There are two cases to be considered when defining
$\tau_{c}(1)$, depending on whether the result operator was in region $r$
before materialization or not. If the result operator was in region $r$, then
the result operator will be part of all the regions in
$\\{r_{c}^{2},r_{c}^{3},\cdots\\}$. Thus, these regions need to execute only
partially for the first tuple to be produced from the result operator. On the
other hand, if the result operator was not in $r$, then it will not be in any
of the newly created regions. Thus, all the newly created regions will have to
execute completely before the region with the result operator starts
executing. Thus, we define $\tau_{c}(1)$ as the following:

$\tau_{c}(1)=\begin{cases}T+t(r_{c}^{1})+min(t(r_{c}^{2}(1)),t(r_{c}^{3}(1)),\cdots),&\text{
result operator}\in r\\\
T+t(r_{c}^{1})+t(r_{c}^{2})+t(r_{c}^{3})+\cdots,&\text{ result operator}\notin
r,\end{cases}$

where $T$ is a constant that denotes the time spent on regions other than $r$,
which are not affected by materialization. The chosen materialization option
is the one that has the smallest estimated $\tau_{c}(1)$.

###  22 Experiments

We performed experiments to show the presence of materialization choices in
real workflows. We also performed experiments to showcase the effect of
materialization choice on the first response time and the size of materialized
data.

####  22.1 Materialization choices in workflows

We analyzed $21$ workflows to see if they needed any materialization. For the
workflows that needed materialization, we also evaluated the number of
materialization choices. We did the analysis on publicly available workflows
from three commercial workflow-based data processing systems, namely Alteryx,
RapidMiner, and Dataiku, as well as workflows in our Texera system. Sample
workflows from the systems are shown in Figures 60, 61, 62, and 63. For each
workflow, we collected results about the need for materialization and the
number of materialization choices. The results are shown in Table 3.

Figure 60:  An Alteryx workflow. Figure 61:  A RapidMiner workflow. Figure 62:  A Dataiku workflow. Figure 63:  A Texera workflow. System |  # of operators |  Materialization needed? |  # of materialization choices
---|---|---|---
Alteryx |  9 |  Y |  3
|  10 |  Y |  3
|  14 |  Y |  4
|  19 |  Y |  1
|  26 |  Y |  12
|  34 |  Y |  1
|  102 |  Y |  1
RapidMiner |  5 |  N |  0
|  9 |  Y |  3
|  10 |  Y |  2
|  10 |  Y |  3
|  13 |  Y |  2
|  19 |  Y |  4
Dataiku |  31 |  Y |  4
|  32 |  Y |  1
|  45 |  N |  0
Texera |  8 |  Y |  1
|  8 |  Y |  1
|  9 |  N |  0
|  10 |  N |  0
|  30 |  Y |  3
Table 3: Analyzing the workflows from various workflow processing systems.

We found a number of workflows from various systems that needed
materialization. Upon examination, many of these workflows offered more than
one materialization choice. The highest number of choices was found in an
Alteryx workflow that offered $12$ choices. The analysis shows that there are
many publicly available workflows that need materialization.

####  22.2 Effect of materialization choice on first response time

We took two workflows of similar structure and showed the effect of
materialization choice on the first response time of the workflows.

Data and workflows. We used two workflows as shown in the Figure 64. Each
workflow used two different sets of tweets as input datasets. Workflow $W1$
used $5$ datasets ranging from $0.3$K tweets to $0.7$K tweets. Workflow $W2$
used $5$ datasets ranging from $1.6$K tweets to $7.6$K tweets. The datasets
used were small because both the workflows contained expensive machine-
learning based operators that took considerable time to process each tweet.
Workflow $W1$ processes the tweets using two machine-learning based operators
that add labels to the tweets specifying whether the text of a tweet is
related to climate change and whether the text suggests action to be taken
regarding climate change. The outputs of the two operators are then joined on
the id column to create a single tuple containing both labels for a tweet.
Workflow $W2$ finds the tweets related to the opioid tramadol. These tweets
are then joined with their reply-tweets using the in-reply-to-id attribute.
Each tuple output of the HashJoin operator contains a tweet about tramadol and
one of its reply tweets. The sentiment label of the reply-tweet is also
present that has been added by the Sentiment Analysis operator. The output of
HashJoin is plotted on a spatial graph with edges between a tweet and its
reply. The reply-tweet location is colored according to its sentiment.

Figure 64:  Workflows used in the experiment to analyze the effect of
materialization choices on the first response time. The blue triangles
represent the materialization choices in the workflows.

Experimental Setting. The experiments were carried out on GCP on a single
n2-highcpu-32 machine containing $32$ CPU cores, $32$ GB RAM and $100$ GB HDD
disk running Ubuntu 18.04.5 LTS. All the operators were given $2$ workers
except the machine-learning based UDF operators in $W1$ which were given $4$
workers because they were expensive.

Materialization choices of W1. The workflow $W1$ had two materialization
choices shown in the Figure 64. The first response times for the two choices
for various input data sizes are shown in Figure 65. The first-response time
when choice $1$ was used for the $0.3$K tweets dataset was $203$ seconds. The
first-response time when choice $2$ was used for the $0.3$K tweets dataset was
$592$ seconds. The first response times gradually increased as the input data
size increased. For the $0.7$K tweets dataset, the first response times for
the choices $1$ and $2$ were $390$ seconds and $1260$ seconds, respectively.
The figure shows that the choice $1$ was consistently better than choice $2$.
The reason is that when choice $2$ was used, the two expensive machine-
learning based operators belonged to the same region and executed
simultaneously. This led to competition between them for resources and slowed
down computation.

Figure 65:  First response time for different input data sizes in $W1$.

Materialization choices of W2. The workflow $W2$ also had two materialization
choices. The first response times for the two choices for various input data
sizes are shown in Figure 66. The selectivity of the Regex operator was low
because only a few tweets contained tramadol. The first-response time when
choice $1$ was used for the $1.6$K tweets dataset was $180$ seconds. The
first-response time when choice $2$ was used for the $1.6$K tweets dataset was
$167$ seconds. The first response times gradually increased as the input data
size increased. For the $7.6$K tweets dataset, the first response times for
the choices $1$ and $2$ were $805$ and $767$ seconds, respectively. The figure
shows that the choice $2$ was consistently better than choice $1$. The reason
for the difference in first-response time is explained next. When choice $1$
was used, the Regex operator belonged to a region that started execution after
the sentiment analysis processed the entire data. Since the Regex operator had
low selectivity, it produced the first output after a considerable time. When
choice $2$ was used, Regex and Sentiment Analysis operators belonged to the
same region and data was processed by the Regex operator alongside Sentiment
Analysis without adding much overhead. When the next region began execution,
the first tuple was immediately output from HashJoin.

Figure 66:  First response time for different input data sizes in $W2$.

Thus, we got lower first-response time when using choice $1$ for $W1$ and
choice $2$ for $W2$. This experiment shows that the materialization choices do
play a role in the first-response time.

####  22.3 Effect of materialization choice on materialized data size

We took the two workflows from the previous experiment and performed
experiments to compare the materialized data size for the various
materialization options. The experiment setting was the same as the previous
experiment.

Figure 67:  Materialization size for different input sizes in $W1$.

The materialization sizes for the two choices in $W_{1}$ for various input
data sizes are shown in Figure 67. In the case of $W_{1}$, the ML2 operator
produced a single tuple for every input tuple. Thus, both choices $1$ and $2$
resulted in the same materialization size consisting of all input tuples. In
the case of $W_{2}$ (Figure 68), the Regex operator was very selective. Thus,
the choice $2$ materialized few tuples, whereas the choice $1$ materialized
all the tuples coming out of Scan.

Figure 68:  Materialization size for different input sizes in $W2$.

A good heuristic while choosing the materialization choices can be to add
materialization after operators that filter out data so that the size of
materialized data is reduced.

###  23 Conclusion

In this chapter, we presented a scheduling framework called Maestro. We showed
how it divides a workflow into regions that can be separately scheduled. We
described an algorithm to create a region graph that encapsulates the region
dependencies. We showed that the the region graph can only be scheduled when
it is acyclic and discussed examples where the region graph has cycles. We
discussed how to use materialization to create an acyclic region graph from a
cyclic region graph. We showed how to enumerate various choices for
materialization. We introduced first response time as a metric of result
awareness and discussed how the materialization choices differ from the
perspective of result awareness. We presented experiments that showed the
presence of materialization choices in real workflows and showcased the effect
of materialization choice on the first-response time and materialized data
size.

##  Chapter 5 Conclusions and Future Work

In this section, we present the conclusions of the three works presented in
this thesis and motivate the future work.

###  24 Conclusions

In this thesis, we presented three works that make data analytics more
interactive, adaptive, and result aware.

In Chapter 2, we presented a system called Amber that supports powerful and
responsive debugging during the execution of a dataflow. It serves as the
backend engine for the Texera service being developed at UC Irvine. We
presented its overall system architecture based on the actor model and
described the whole lifecycle of the execution of a workflow, including how
control messages are sent to actors, how to expedite the processing of these
control messages, and how to pause and resume the computation of each actor.
We studied how to support conditional breakpoints, and presented solutions for
enforcing local conditional breakpoints and global conditional breakpoints. We
developed a technique to support fault tolerance in Amber, which is
challenging due to the presence of control messages. We implemented Amber on
top of Orleans, and presented an extensive experimental evaluation to show its
high usability and performance comparable to Spark.

In Chapter 3, we presented a framework called Reshape that adaptively handles
partitioning skew in the exploratory data analysis setting. We presented
different approaches for load transfer and analyzed their impact on the
results shown to the user. We presented an analysis about the effect of the
skew-detection threshold on mitigation and used it to adaptively adjust the
threshold. We generalized Reshape to other operators such as HashJoin, Sort,
and GroupBy, and broader execution settings. We implemented Reshape on top of
two big data engines - Amber and Flink - and presented an experimental
evaluation.

In Chapter 4, we presented a scheduling framework called Maestro. We showed
how it divides a workflow into regions that can be separately scheduled. We
described an algorithm to create a region graph that are based on the region
dependencies. Since the region graph can only be scheduled when it is acyclic,
we need to avoid cycles in the region graph. We discussed how to use
materialization to create an acyclic region graph. We showed how to enumerate
various choices for materialization. We introduced first response time as a
metric of result awareness and discussed how the materialization choices
differ from the perspective of result awareness. We presented experiments that
showed the presence of materialization choices in real workflows and showcased
the effect of materialization choice on the first-response time and
materialized data size.

###  25 Future Work

Amber. The determination of number of workers per operator and the worker
placement in Amber is static. This can be improved to allow adaptation in the
number of workers and their placement depending on the complexity of the
operators and the data traffic pattern among machines. The global breakpoint
detection in Amber is not deterministic, in the sense that two executions may
result in different numbers of tuples being processed by the workers before
the breakpoint is hit. It would be interesting to devise a technique which
leads to deterministic breakpoint detection and analyze the overhead of this
technique on the total processing time.

Reshape. The current framework of Reshape focuses on skew handling among the
workers of an operator. There can be multiple skewed operators in a workflow.
It would be interesting to see how Reshape can be extended to consider the
skew in multiple operators and make global decisions. It would also be
interesting to extend Reshape so that a single worker can serve as the helper
for multiple skewed workers. Another future work is using both split by key
and split by records among the workers of an operator, i.e., load transfer for
a few keys happens using split by key and for other keys using split by
records. The chapter does not consider the existence of windows on the input
data. When extending Reshape to work on infinite streams, it is important to
consider how the existence of windows will affect the decision of Reshape.

Maestro. The current definition of first-response time in Maestro and the
process of choosing a materialization option assumes a single sink. These
concepts have to be extended to the case of multiple sinks. The algorithm to
run an exhaustive search over the multiple materialization choices may be
expensive. We need a way, maybe using heuristics, to prune the search space.

## References

  * [1] Cloudberry - Big Data Visualization.  http://cloudberry.ics.uci.edu/, 2018.
  * [2] A. S. Abdelhamid, A. R. Mahmood, A. Daghistani, and W. G. Aref.  Prompt: Dynamic data-partitioning for distributed micro-batch stream processing systems.  In D. Maier, R. Pottinger, A. Doan, W. Tan, A. Alawini, and H. Q. Ngo, editors, Proceedings of the 2020 International Conference on Management of Data, SIGMOD Conference 2020, online conference [Portland, OR, USA], June 14-19, 2020, pages 2455–2469. ACM, 2020.
  * [3] G. A. Agha.  Actors: A model of concurrent computation in distributed systems.  Technical report, Massachusetts Inst Of Tech Cambridge Artificial Intelligence Lab, 1985.
  * [4] Akka Website, https://akka.io/.
  * [5] S. Alsubaiee, Y. Altowim, H. Altwaijry, A. Behm, V. R. Borkar, Y. Bu, M. J. Carey, I. Cetindil, M. Cheelangi, K. Faraaz, E. Gabrielova, R. Grover, Z. Heilbron, Y. Kim, C. Li, G. Li, J. M. Ok, N. Onose, P. Pirzadeh, V. J. Tsotras, R. Vernica, J. Wen, and T. Westmann.  Asterixdb: A scalable, open source BDMS.  CoRR, abs/1407.0454, 2014.
  * [6] Alteryx Website, https://www.alteryx.com/.
  * [7] L. Aniello, R. Baldoni, and L. Querzoni.  Adaptive online scheduling in storm.  In S. Chakravarthy, S. D. Urban, P. R. Pietzuch, and E. A. Rundensteiner, editors, The 7th ACM International Conference on Distributed Event-Based Systems, DEBS ’13, Arlington, TX, USA - June 29 - July 03, 2013, pages 207–218. ACM, 2013.
  * [8] P. Antonopoulos, H. Kodavalla, A. Tran, N. Upreti, C. Shah, and M. Sztajno.  Resumable online index rebuild in SQL server.  PVLDB, 10(12):1742–1753, 2017.
  * [9] Apache AsterixDB, http://asterixdb.apache.org.
  * [10] Apache Flink http://flink.apache.org.
  * [11] Apache Hadoop MapReduce, https://hadoop.apache.org/docs/stable/hadoop-mapreduce-client/hadoop-mapreduce-client-core/MapReduceTutorial.html.
  * [12] Apache samza.  http://samza.apache.org/.
  * [13] Apache Spark http://spark.apache.org.
  * [14] Apache Storm, http://storm.apache.org/.
  * [15] M. Armbrust, T. Das, J. Torres, B. Yavuz, S. Zhu, R. Xin, A. Ghodsi, I. Stoica, and M. Zaharia.  Structured streaming: A declarative API for real-time applications in apache spark.  In G. Das, C. M. Jermaine, and P. A. Bernstein, editors, Proceedings of the 2018 International Conference on Management of Data, SIGMOD Conference 2018, Houston, TX, USA, June 10-15, 2018, pages 601–613. ACM, 2018.
  * [16] P. Auer, N. Cesa-Bianchi, and P. Fischer.  Finite-time analysis of the multiarmed bandit problem.  Mach. Learn., 47(2-3):235–256, 2002.
  * [17] Handling Backpressure https://medium.com/@jayphelps/backpressure-explained-the-flow-of-data-through-software-2350b3e77ce7.
  * [18] L. Battle, D. Fisher, R. DeLine, M. Barnett, B. Chandramouli, and J. Goldstein.  Making sense of temporal queries with interactive visualization.  In Proceedings of the 2016 CHI Conference on Human Factors in Computing Systems, San Jose, CA, USA, May 7-12, 2016, pages 5433–5443, 2016\.
  * [19] E. Begoli, T. Akidau, S. Chernyak, F. Hueske, K. Knight, K. Knowles, D. Mills, and D. Sotolongo.  Watermarks in stream processing systems: Semantics and comparative analysis of apache flink and google cloud dataflow.  Proc. VLDB Endow., 14(12):3135–3147, 2021.
  * [20] A. Behm, S. Palkar, U. Agarwal, T. Armstrong, D. Cashman, A. Dave, T. Greenstein, S. Hovsepian, R. Johnson, A. S. Krishnan, P. Leventis, A. Luszczak, P. Menon, M. Mokhtar, G. Pang, S. Paranjpye, G. Rahn, B. Samwel, T. van Bussel, H. V. Hovell, M. Xue, R. Xin, and M. Zaharia.  Photon: A fast query engine for lakehouse systems.  In Z. Ives, A. Bonifati, and A. E. Abbadi, editors, SIGMOD ’22: International Conference on Management of Data, Philadelphia, PA, USA, June 12 - 17, 2022, pages 2326–2339. ACM, 2022.
  * [21] S. M. Beitzel, E. C. Jensen, A. Chowdhury, D. A. Grossman, and O. Frieder.  Hourly analysis of a very large topically categorized web query log.  In M. Sanderson, K. Järvelin, J. Allan, and P. Bruza, editors, SIGIR 2004: Proceedings of the 27th Annual International ACM SIGIR Conference on Research and Development in Information Retrieval, Sheffield, UK, July 25-29, 2004, pages 321–328. ACM, 2004.
  * [22] A. Benoit, Ü. V. Çatalyürek, Y. Robert, and E. Saule.  A survey of pipelined workflow scheduling: Models and algorithms.  ACM Comput. Surv., 45(4):50:1–50:36, 2013.
  * [23] P. A. Bernstein, M. Dashti, T. Kiefer, and D. Maier.  Indexing in an actor-oriented database.  In CIDR 2017, 8th Biennial Conference on Innovative Data Systems Research, Chaminade, CA, USA, January 8-11, 2017, Online Proceedings, 2017.
  * [24] I. Beschastnikh, P. Wang, Y. Brun, and M. D. Ernst.  Debugging distributed systems.  Commun. ACM, 59(8):32–37, 2016.
  * [25] L. Bindschaedler, J. Malicevic, N. Schiper, A. Goel, and W. Zwaenepoel.  Rock you like a hurricane: taming skew in large scale analytics.  In R. Oliveira, P. Felber, and Y. C. Hu, editors, Proceedings of the Thirteenth EuroSys Conference, EuroSys 2018, Porto, Portugal, April 23-26, 2018, pages 20:1–20:15. ACM, 2018.
  * [26] V. R. Borkar.  An Efficient Foundation for Big Data Processing on Modern Clusters.  PhD thesis, University of California, Irvine, USA, 2016.
  * [27] V. R. Borkar, M. J. Carey, R. Grover, N. Onose, and R. Vernica.  Hyracks: A flexible and extensible foundation for data-intensive computing.  In International Conference on Data Engineering, pages 1151–1162, 2011.
  * [28] CAF Website, https://actor-framework.org/.
  * [29] P. Carbone, S. Ewen, G. Fóra, S. Haridi, S. Richter, and K. Tzoumas.  State management in apache flink®: Consistent stateful distributed stream processing.  Proc. VLDB Endow., 10(12):1718–1729, 2017.
  * [30] P. Carbone, G. Fóra, S. Ewen, S. Haridi, and K. Tzoumas.  Lightweight asynchronous snapshots for distributed dataflows.  CoRR, abs/1506.08603, 2015.
  * [31] P. Carbone, M. Fragkoulis, V. Kalavri, and A. Katsifodimos.  Beyond analytics: The evolution of stream processing systems.  In D. Maier, R. Pottinger, A. Doan, W. Tan, A. Alawini, and H. Q. Ngo, editors, Proceedings of the 2020 International Conference on Management of Data, SIGMOD Conference 2020, online conference [Portland, OR, USA], June 14-19, 2020, pages 2651–2658. ACM, 2020.
  * [32] B. Chandramouli, C. N. Bond, S. Babu, and J. Yang.  Query suspend and resume.  In Proceedings of the ACM SIGMOD International Conference on Management of Data, Beijing, China, June 12-14, 2007, pages 557–568, 2007.
  * [33] K. M. Chandy and L. Lamport.  Distributed snapshots: Determining global states of distributed systems.  ACM Trans. Comput. Syst., 3(1):63–75, 1985.
  * [34] S. Chaudhuri, R. Motwani, and V. R. Narasayya.  Random sampling for histogram construction: How much is enough?  In L. M. Haas and A. Tiwary, editors, SIGMOD 1998, Proceedings ACM SIGMOD International Conference on Management of Data, June 2-4, 1998, Seattle, Washington, USA, pages 436–447. ACM Press, 1998.
  * [35] Q. Chen, J. Yao, and Z. Xiao.  LIBRA: lightweight data skew mitigation in mapreduce.  IEEE Trans. Parallel Distributed Syst., 26(9):2520–2533, 2015\.
  * [36] D. Cheng, Y. Chen, X. Zhou, D. Gmach, and D. S. Milojicic.  Adaptive scheduling of parallel jobs in spark streaming.  In 2017 IEEE Conference on Computer Communications, INFOCOM 2017, Atlanta, GA, USA, May 1-4, 2017, pages 1–9. IEEE, 2017.
  * [37] D. Dao, J. Albrecht, C. Killian, and A. Vahdat.  Live debugging of distributed systems.  In International Conference on Compiler Construction, pages 94–108. Springer, 2009.
  * [38] J. Dean and S. Ghemawat.  Mapreduce: Simplified data processing on large clusters.  In OSDI, pages 137–150, 2004.
  * [39] D. J. DeWitt, S. Ghandeharizadeh, D. A. Schneider, A. Bricker, H.-I. Hsiao, and R. Rasmussen.  The Gamma database machine project.  IEEE Trans. Knowl. Data Eng., 2(1):44–62, 1990.
  * [40] D. J. DeWitt and J. Gray.  Parallel database systems: The future of high performance database systems.  Commun. ACM, 35(6):85–98, 1992.
  * [41] D. J. DeWitt, J. F. Naughton, D. A. Schneider, and S. Seshadri.  Practical skew handling in parallel joins.  In L. Yuan, editor, 18th International Conference on Very Large Data Bases, August 23-27, 1992, Vancouver, Canada, Proceedings, pages 27–40. Morgan Kaufmann, 1992.
  * [42] S. Di, D. Kondo, and W. Cirne.  Host load prediction in a google compute cloud with a bayesian model.  In J. K. Hollingsworth, editor, SC Conference on High Performance Computing Networking, Storage and Analysis, SC ’12, Salt Lake City, UT, USA - November 11 - 15, 2012, page 21. IEEE/ACM, 2012.
  * [43] B. Ding, S. Chaudhuri, J. Gehrke, and V. R. Narasayya.  DSB: A decision support benchmark for workload-driven and traditional database systems.  Proc. VLDB Endow., 14(13):3376–3388, 2021.
  * [44] J. Ding, T. Z. J. Fu, R. T. B. Ma, M. Winslett, Y. Yang, Z. Zhang, and H. Chao.  Optimal operator state migration for elastic data stream processing.  CoRR, abs/1501.03619, 2015.
  * [45] Einblick, https://www.einblick.ai.
  * [46] M. Elseidy, A. Elguindy, A. Vitorovic, and C. Koch.  Scalable and adaptive online joins.  Proc. VLDB Endow., 7(6):441–452, 2014.
  * [47] P. Eng, B. C. Ooi, H. S. Sim, and K. Tan.  Preference-driven query processing.  In U. Dayal, K. Ramamritham, and T. M. Vijayaraman, editors, Proceedings of the 19th International Conference on Data Engineering, March 5-8, 2003, Bangalore, India, pages 671–673. IEEE Computer Society, 2003.
  * [48] R. C. Fernandez, M. Migliavacca, E. Kalyvianaki, and P. R. Pietzuch.  Integrating scale out and fault tolerance in stream processing using operator state management.  In K. A. Ross, D. Srivastava, and D. Papadias, editors, Proceedings of the ACM SIGMOD International Conference on Management of Data, SIGMOD 2013, New York, NY, USA, June 22-27, 2013, pages 725–736. ACM, 2013.
  * [49] D. Fisher, R. DeLine, M. Czerwinski, and S. M. Drucker.  Interactions with big data analytics.  Interactions, 19(3):50–59, 2012.
  * [50] Apache flink pipelined region scheduling.  https://flink.apache.org/2020/12/15/pipelined-region-sheduling.html.
  * [51] J. Fowler and W. Zwaenepoel.  Causal distributed breakpoints.  In 10th International Conference on Distributed Computing Systems (ICDCS 1990), May 28 - June 1, 1990, Paris, France, pages 134–141, 1990.
  * [52] D. R. Fulkerson and L. R. Ford.  Flows in networks.  Princeton University Press Princeton, 1962.
  * [53] P. Garraghan, X. Ouyang, P. Townend, and J. Xu.  Timely long tail identification through agent based monitoring and analytics.  In IEEE 18th International Symposium on Real-Time Distributed Computing, ISORC 2015, Auckland, New Zealand, 13-17 April, 2015, pages 19–26. IEEE Computer Society, 2015.
  * [54] G. Graefe.  Query evaluation techniques for large databases.  ACM Comput. Surv., 25(2):73–170, 1993.
  * [55] B. Gufler, N. Augsten, A. Reiser, and A. Kemper.  Load balancing in mapreduce based on scalable cardinality estimates.  In A. Kementsietsidis and M. A. V. Salles, editors, IEEE 28th International Conference on Data Engineering (ICDE 2012), Washington, DC, USA (Arlington, Virginia), 1-5 April, 2012, pages 522–533. IEEE Computer Society, 2012.
  * [56] M. A. Gulzar, M. Interlandi, T. Condie, and M. Kim.  Debugging big data analytics in spark with _BigDebug_.  In Proceedings of the 2017 ACM International Conference on Management of Data, SIGMOD Conference 2017, Chicago, IL, USA, May 14-19, 2017, pages 1627–1630, 2017.
  * [57] M. A. Gulzar, M. Interlandi, X. Han, M. Li, T. Condie, and M. Kim.  Automated debugging in data-intensive scalable computing.  In Proceedings of the 2017 Symposium on Cloud Computing, SoCC 2017, Santa Clara, CA, USA, September 24-27, 2017, pages 520–534, 2017.
  * [58] C. Gupta, A. Mehta, and U. Dayal.  PQR: predicting query execution times for autonomous workload management.  In J. Strassner, S. A. Dobson, J. A. B. Fortes, and K. K. Goswami, editors, 2008 International Conference on Autonomic Computing, ICAC 2008, June 2-6, 2008, Chicago, Illinois, USA, pages 13–22. IEEE Computer Society, 2008.
  * [59] D. Haban and W. Weigel.  Global events and global breakpoints in distributed systems.  In [1988] Proceedings of the Twenty-First Annual Hawaii International Conference on System Sciences. Volume II: Software track, volume 2, pages 166–175. IEEE, 1988.
  * [60] Apache hadoop mapreduce configuration parameters.  https://hadoop.apache.org/docs/stable/hadoop-mapreduce-client/hadoop-mapreduce-client-core/mapred-default.xml.
  * [61] Halo Website, https://www.halowaypoint.com/en-us.
  * [62] C. Herath, F. Liu, S. Marru, L. Gunathilake, M. Sosonkina, J. P. Vary, P. Maris, and M. E. Pierce.  Web service andworkflow abstractions to large scale nuclear physics calculations.  In 2012 IEEE Ninth International Conference on Services Computing, Honolulu, HI, USA, June 24-29, 2012, pages 703–710, 2012.
  * [63] C. Hewitt, P. B. Bishop, and R. Steiger.  A universal modular ACTOR formalism for artificial intelligence.  In Proceedings of the 3rd International Joint Conference on Artificial Intelligence. Standford, CA, USA, August 20-23, 1973, pages 235–245, 1973.
  * [64] B. Hindman, A. Konwinski, M. Zaharia, A. Ghodsi, A. D. Joseph, R. H. Katz, S. Shenker, and I. Stoica.  Mesos: A platform for fine-grained resource sharing in the data center.  In D. G. Andersen and S. Ratnasamy, editors, Proceedings of the 8th USENIX Symposium on Networked Systems Design and Implementation, NSDI 2011, Boston, MA, USA, March 30 - April 1, 2011. USENIX Association, 2011.
  * [65] M. Hoffmann, A. Lattuada, F. McSherry, V. Kalavri, J. Liagouris, and T. Roscoe.  Megaphone: Latency-conscious state migration for distributed streaming dataflows.  Proc. VLDB Endow., 12(9):1002–1015, 2019.
  * [66] K. Hsieh, G. Ananthanarayanan, P. Bodík, S. Venkataraman, P. Bahl, M. Philipose, P. B. Gibbons, and O. Mutlu.  Focus: Querying large video datasets with low latency and low cost.  In A. C. Arpaci-Dusseau and G. Voelker, editors, 13th USENIX Symposium on Operating Systems Design and Implementation, OSDI 2018, Carlsbad, CA, USA, October 8-10, 2018, pages 269–286. USENIX Association, 2018\.
  * [67] M. Interlandi, K. Shah, S. D. Tetali, M. A. Gulzar, S. Yoo, M. Kim, T. D. Millstein, and T. Condie.  Titian: Data provenance support in spark.  PVLDB, 9(3):216–227, 2015.
  * [68] R. J. B. Jr. and D. P. Miranker.  Processing queries for first few answers.  In CIKM ’96, Proceedings of the Fifth International Conference on Information and Knowledge Management, November 12 - 16, 1996, Rockville, Maryland, USA, pages 45–52. ACM, 1996.
  * [69] I. K. Kim, W. Wang, Y. Qi, and M. Humphrey.  Empirical evaluation of workload forecasting techniques for predictive cloud resource scaling.  In 9th IEEE International Conference on Cloud Computing, CLOUD 2016, San Francisco, CA, USA, June 27 - July 2, 2016, pages 1–10. IEEE Computer Society, 2016.
  * [70] Knime Website, https://www.knime.com/.
  * [71] D. Kossmann.  The state of the art in distributed query processing.  ACM Comput. Surv., 32(4):422–469, 2000.
  * [72] A. Kulkarni, J. Teevan, K. M. Svore, and S. T. Dumais.  Understanding temporal query dynamics.  In I. King, W. Nejdl, and H. Li, editors, Proceedings of the Forth International Conference on Web Search and Web Data Mining, WSDM 2011, Hong Kong, China, February 9-12, 2011, pages 167–176. ACM, 2011.
  * [73] A. Kumar, Z. Wang, S. Ni, and C. Li.  Amber: A debuggable dataflow system based on the actor model.  Proc. VLDB Endow., 13(5):740–753, 2020.
  * [74] Y. Kwon, M. Balazinska, B. Howe, and J. A. Rolia.  Skewtune: mitigating skew in mapreduce applications.  In K. S. Candan, Y. Chen, R. T. Snodgrass, L. Gravano, and A. Fuxman, editors, Proceedings of the ACM SIGMOD International Conference on Management of Data, SIGMOD 2012, Scottsdale, AZ, USA, May 20-24, 2012, pages 25–36. ACM, 2012.
  * [75] R. Lawrence.  Early hash join: A configurable algorithm for the efficient and early production of join results.  In K. Böhm, C. S. Jensen, L. M. Haas, M. L. Kersten, P. Larson, and B. C. Ooi, editors, Proceedings of the 31st International Conference on Very Large Data Bases, Trondheim, Norway, August 30 - September 2, 2005, pages 841–852. ACM, 2005.
  * [76] X. Liu, Z. Wang, S. Ni, S. Alsudais, Y. Huang, A. Kumar, and C. Li.  Demonstration of collaborative and interactive workflow-based data analytics in texera.  Proc. VLDB Endow., 15(12):3738–3741, 2022.
  * [77] B. Ludäscher, I. Altintas, C. Berkley, D. Higgins, E. Jaeger, M. Jones, E. A. Lee, J. Tao, and Y. Zhao.  Scientific workflow management and the kepler system.  Concurrency and Computation: Practice and Experience, 18(10):1039–1065, 2006.
  * [78] L. Mai, K. Zeng, R. Potharaju, L. Xu, S. Suh, S. Venkataraman, P. Costa, T. Kim, S. Muthukrishnan, V. Kuppa, S. Dhulipalla, and S. Rao.  Chi: A scalable and programmable control plane for distributed stream processing systems.  Proc. VLDB Endow., 11(10):1303–1316, 2018.
  * [79] H. Mao, M. Schwarzkopf, S. B. Venkatakrishnan, Z. Meng, and M. Alizadeh.  Learning scheduling algorithms for data processing clusters.  In J. Wu and W. Hall, editors, Proceedings of the ACM Special Interest Group on Data Communication, SIGCOMM 2019, Beijing, China, August 19-23, 2019, pages 270–288. ACM, 2019.
  * [80] S. Marru, L. Gunathilake, C. Herath, P. Tangchaisin, M. E. Pierce, C. Mattmann, R. Singh, T. Gunarathne, E. Chinthaka, R. Gardler, A. Slominski, A. Douma, S. Perera, and S. Weerawarana.  Apache airavata: a framework for distributed applications and computational workflows.  In Proceedings of the 2011 ACM SC Workshop on Gateway Computing Environments, GCE 2011, Seattle, WA, USA, November 18, 2011, pages 21–28, 2011.
  * [81] A. Metwally, D. Agrawal, and A. E. Abbadi.  Efficient computation of frequent and top-k elements in data streams.  In T. Eiter and L. Libkin, editors, Database Theory - ICDT 2005, 10th International Conference, Edinburgh, UK, January 5-7, 2005, Proceedings, volume 3363 of Lecture Notes in Computer Science, pages 398–412. Springer, 2005.
  * [82] B. P. Miller and J. Choi.  Breakpoints and halting in distributed programs.  In Proceedings of the 8th International Conference on Distributed Computing Systems, San Jose, California, USA, June 13-17, 1988, pages 316–323, 1988.
  * [83] P. Missier, S. Soiland-Reyes, S. Owen, W. Tan, A. Nenadic, I. Dunlop, A. Williams, T. Oinn, and C. Goble.  Taverna, reloaded.  In International conference on scientific and statistical database management, pages 471–481. Springer, 2010.
  * [84] B. D. Monte, S. Zeuch, T. Rabl, and V. Markl.  Rhino: Efficient management of very large distributed state for stream processing engines.  In D. Maier, R. Pottinger, A. Doan, W. Tan, A. Alawini, and H. Q. Ngo, editors, Proceedings of the 2020 International Conference on Management of Data, SIGMOD Conference 2020, online conference [Portland, OR, USA], June 14-19, 2020, pages 2471–2486. ACM, 2020.
  * [85] D. Moritz, D. Halperin, B. Howe, and J. Heer.  Perfopticon: Visual query analysis for distributed databases.  Comput. Graph. Forum, 34(3):71–80, 2015.
  * [86] M. A. U. Nasir, G. D. F. Morales, D. García-Soriano, N. Kourtellis, and M. Serafini.  The power of both choices: Practical load balancing for distributed stream processing engines.  In J. Gehrke, W. Lehner, K. Shim, S. K. Cha, and G. M. Lohman, editors, 31st IEEE International Conference on Data Engineering, ICDE 2015, Seoul, South Korea, April 13-17, 2015, pages 137–148. IEEE Computer Society, 2015.
  * [87] M. A. U. Nasir, G. D. F. Morales, N. Kourtellis, and M. Serafini.  When two choices are not enough: Balancing at scale in distributed stream processing.  In 32nd IEEE International Conference on Data Engineering, ICDE 2016, Helsinki, Finland, May 16-20, 2016, pages 589–600. IEEE Computer Society, 2016.
  * [88] L. Neumeyer, B. Robbins, A. Nair, and A. Kesari.  S4: distributed stream computing platform.  In ICDMW 2010, The 10th IEEE International Conference on Data Mining Workshops, Sydney, Australia, 13 December 2010, pages 170–177, 2010\.
  * [89] For big-data scientists, ’janitor work’ is key hurdle to insights.  https://www.nytimes.com/2014/08/18/technology/for-big-data-scientists-hurdle-to-insights-is-janitor-work.html.
  * [90] C. Olston and B. Reed.  Inspector gadget: a framework for custom monitoring and debugging of distributed dataflows.  In Proceedings of the ACM SIGMOD International Conference on Management of Data, SIGMOD 2011, Athens, Greece, June 12-16, 2011, pages 1221–1224, 2011.
  * [91] Orleans Website, https://dotnet.github.io/orleans/.
  * [92] M. T. Özsu and P. Valduriez.  Principles of Distributed Database Systems, 4th Edition.  Springer, 2020.
  * [93] B. Peng, M. Hosseini, Z. Hong, R. Farivar, and R. H. Campbell.  R-storm: Resource-aware scheduling in storm.  In R. Lea, S. Gopalakrishnan, E. Tilevich, A. L. Murphy, and M. Blackstock, editors, Proceedings of the 16th Annual Middleware Conference, Vancouver, BC, Canada, December 07 - 11, 2015, pages 149–161. ACM, 2015.
  * [94] A. D. Popescu, V. Ercegovac, A. Balmin, M. Branco, and A. Ailamaki.  Same queries, different data: Can we predict runtime performance?  In A. Kementsietsidis and M. A. V. Salles, editors, Workshops Proceedings of the IEEE 28th International Conference on Data Engineering, ICDE 2012, Arlington, VA, USA, April 1-5, 2012, pages 275–280. IEEE Computer Society, 2012.
  * [95] Prediction interval, https://otexts.com/fpp2/prediction-intervals.html.
  * [96] ProtoActor Website, http://proto.actor.
  * [97] Ptolemy II Website, https://ptolemy.berkeley.edu/ptolemyII/ptII11.0/index.htm.
  * [98] S. R. Ramakrishnan, G. Swart, and A. Urmanov.  Balancing reducer skew in mapreduce workloads using progressive sampling.  In M. J. Carey and S. Hand, editors, ACM Symposium on Cloud Computing, SOCC ’12, San Jose, CA, USA, October 14-17, 2012, page 16. ACM, 2012.
  * [99] RapidMiner Website, https://rapidminer.com/.