Efficient resource scheduling for the analysis of Big Data streams

Abstract

The emergence of Big Data has had a profound impact on how data are analyzed. Open source distributed stream processing platforms have gained popularity for analyzing streaming Big Data as they provide low latency required for streaming Big Data applications using cluster resources. However, existing resource schedulers are still lacking the efficiency that Big Data analytical applications require. Recent works have already considered streaming Big Data characteristics to improve the efficiency of scheduling in the platforms. Nevertheless, they have not taken into account the specific attributes of analytical applications. This study, therefore, presents Bframework, an efficient resource scheduling framework used by streaming Big Data analysis applications based on cluster resources. Bframework proposes a query model using Directed Graphs (DGs) and introduces operator assignment and operator scheduling algorithms based on a novel partitioning algorithm. Bframework is highly adaptable to the fluctuation of streaming Big Data and the availability of cluster resources. Experiments with the benchmark and well-known real-world queries show that Bframework can significantly reduce the latency of streaming Big Data analysis queries up to about 65%.

Keywords

Streaming Big Data analytical query resource scheduling distributed stream processing

1. Introduction

The emergence of data-intensive applications has rapidly increased the volume, variety and velocity of the generated data [14]. Efficient computing of the generated data during its lifecycle is beyond the capability of the current hardware and software technologies [11]. This represents a major challenge for many organizations and is known as the Big Data problem [13].

Streaming Big Data is related to the velocity dimension of Big Data and refers both to how fast data are generated and how fast they need to be analyzed [3]. Analysis of streaming Big Data is the last and most important stage of the streaming Big Data lifecycle in analytical applications [9]. The real-world examples of the applications can be analysis of networks traffic, healthcare data, financial transactions or web-clicks [11]. To develop such applications; high-level declarative query languages like SQL are usually preferred over coding them directly in algorithmic languages such as Java [12]. Therefore, most of the open-source distributed stream computing platforms (e.g., Storm,1

¹
storm.apache.org.

Heron2

https://twitter.github.io/heron/.

and Flink3

https://flink.apache.org.

platforms) also support declarative languages to develop and execute analysis queries on the cluster of computing nodes [11]. Efficient execution of such queries is considerably affected by scheduling decisions made by the platforms.

Current open source Big Data stream computing platforms do not adequately address the problem of efficiently scheduling analytical queries. For example, Storm platform uses a round-robin strategy that is not aware of tasks or resources characteristics (i.e., task structure, data stream traffic of task) or the resource availability. Recent studies have proposed solutions to improve the efficiency of the platform schedulers by taking into account these aspects. However, most of them only consider one of these aspects or do not achieve satisfactory results.

Our work is motivated by the observation that the existing resource-aware, task-aware or traffic-aware scheduling strategies are inefficient and not suitable for the analytic queries. Current resource-aware works tackle the inefficiency of schedulers to reduce the latency of queries by considering the capability of the resources, but they generally ignore the resource load. Task-aware works reduce the latency of queries by taking into account the structure of queries, but they generally ignore the query runtime resource demands. In addition, traffic-aware works take into account the fluctuation of the data stream, but they do not generally consider all streaming Big Data characteristics.

We developed a scheduler with some lower tuple latency suitable for streaming Big Data analysis by considering the capacity and the load of the resources, the structure and the dynamic resource demand of queries, and the volume, velocity and variety of the input Big Data streams. Our approach is inspired by the idea that using the structure of streaming Big Data analytic query to partition the query operators and applying distinct scheduling strategy for each partition can reduce the average tuple latency of the query. More precisely, as query sample shown in Fig. 1, the weight of the outgoing edge of the operator ${O}_{\textit{Join}}$ ( ${{\triangleright\triangleleft}}_{1}$ ) is considerably higher than the operator ${O}_{\textit{Project}}(P_{1})$ outgoing edge; using this edge weight difference can result in multi partitions of the operators and therefore a specific scheduling strategy can be applied for each partition.

Key contributions of this study can be summarized as follows:

•

Formal definition of the Big Data stream and the Big Data stream analytic query.

•

Systematic determination of the initial number of threads, processes and computing nodes needed for a streaming Big Data analytic query execution.

•

The off-line initial assignment and scheduling of the streaming Big Data analytic query operators to the initial threads, processes and computing nodes using Bframework profiling and partitioning, operator assignment and Off-line Bscheduler algorithms.

•

The online re-scheduling of the streaming Big Data analytic query operators to the new computing nodes using Bframework’s Online Bscheduler.

•

Prototype implementation, simulation and performance evaluation of Bframework.

The remainder of this paper is organized as follows. Section 2 reviews the related works on Big Data and distributed stream processing schedulers. In Section 3, the definitions of Big Data stream and analytic query scheduling model in Big Data stream computing environment are presented. Section 3 presents the system architecture of Bframework. Section 4 focuses on the partitioning, assignment and scheduling algorithms of Bframework. Section 5 shows the experimental environment, parameter setup, and the experimental results of Bframework. Finally, conclusions and future works are given in Section 6.

2. Related works

This section presents two broad categories of related works: scheduling in distributed stream processing (DSP) open source platforms and the scheduling for stream Big Data queries or tasks.

2.1 DSP scheduling

In this section, Bframework as a DSP scheduler is compared with the recent state-of-the-art studies in terms of the proposed task, resource and scheduling models.

In [10], a stream processing system is designed and implemented based on Storm platform whose traffic-aware scheduler assigns and reassigns accepted tasks at runtime. Furthermore, the scheduler aims to minimize the inter-node and inter-process traffic loads and also monitors the overload of computing nodes. In comparison with Bframework, the scheduler is neither resource-aware nor topology-aware because its task model only considers the tasks as a set of given threads and models resources as a set of given processes. Furthermore, its scheduling model only takes into account the inter-node traffic optimization.

In [23], Storm platform is extended to operate in geographically distributed and highly variable environments. Authors present a distributed scheduler which is QoS-aware and operates at runtime. It takes into account the latency, resource utilization and resource availability as QoS attributes. In comparison with Bframework, its scheduling model aims to solve two placement problems, assignment of the task operators to virtual machine and assignment of the virtual machine to the physical machines where in comparison with our work, Bframework only considers physical machines. Furthermore its scheduler can be distributed among multiple worker nodes which the Bframework scheduler is not capable of. In addition, no special strategies are proposed by the authors while Bframework has proposed efficient strategies for thread, process and node assignment. And also, its task models considers tasks as a set of the operators while it is modelled as directed graph in Bframework.

In [21], a predictive scheduling framework is proposed which predicts the processing time of the application based on the structure of the topology. It also presents a scheduling algorithm which assigns threads to machines based on the prediction result. In comparison with Bframework, its task model considers tasks as a set of threads and processes; its resource model considers a set of machines, however their load and capacity is not considered. Similar to Bframework, its scheduling model models the scheduling as two mapping problems, assignment of the thread to process and process to computing node but there is no operator to thread assignment in their study. Compared with Bframework, it only aims to minimize the inter operator traffic load of a task.

In [25, 24], the placement problem for DSP is studied and a scheduler for Storm DSP is presented based on Integer Linear Programming. Similar to Bframework, they model tasks as a graph. Its resource model considers computing nodes with respect to QoS metrics like resource availability but the capacity of computing nodes is ignored. Its objectives model the two QoS metrics of resource availability and response time. It also takes into account the network delay. In comparison with Bframework, workload balancing and distribution is not considered.

In [20], authors aim to provision resources needed for data stream applications in presence of workload fluctuation. They use a control-theoretic method to find an optimal stream application configuration to address latency constraints. In comparison with Bframework, they mainly focus on the resource provisioning problem where the focus of Bframework is on both operator assignment and scheduling problems.

The Flink platform, includes a scheduling framework where process computing units are able to host a layer of accepted tasks and one or multiple layer of tasks can be scheduled in a computing node. As result, its scheduling framework is not QoS-aware. In [18], a scheduler for SPS is proposed using SDN controllers to place tasks to the underlying resources while taking into account the task QoS requirements. In comparison with Bframework, the resource model of s-flink is completely different from Bframework resource model. In [19], a new distributed system for managing stream processing application is developed to overcome the limitations of Storm. In this system, Aurora [6], a generic scheduler framework is used as resource scheduling. In comparison with Bframework, there are no specific strategies are applied.

In [5], a task optimization and task scheduling is proposed. The scheduling algorithm takes into account the topology and traffic of tasks and resources. Its task model takes into account task as Directed Acyclic Graph (DAG) and the scheduling algorithm assigns operator to thread and thread to computing nodes. Its objective is to minimize the inter node traffic. In comparison with Bframework, its task, resource and scheduling model has close assumption with Bframework, but it is not specifically designed for analytic queries and its solution has some limitations.

In summary, among the optimization strategies for improving the DSP scheduling, most studies consider the topology structure, the inter-node traffic or computing node load aspects of the scheduling’s improvement. However, our proposed scheduling algorithm could improve DSP scheduling in all of the mentioned aspects, which jointly considers the topology structure, inter-node traffic and workload of computing nodes.

2.2 Streaming Big Data scheduling

In this section, Bframework as a streaming Big Data scheduler is compared with the recent state-of-the-art studies in terms of the proposed task, resource and scheduling models.

In [11] a real-time scheduling framework is proposed which aims to meet the low response time and high energy efficiency of the query. Similar to Bframework, it models tasks as graph which encompasses operators and their inputs and outputs but the variety dimension of streaming Big Data is not considered. Similar to the Bframework, it models resources by considering resource capacity and the aim of its scheduling model is to minimize the response time and maximize the energy efficiency of the computing nodes. However, workload balancing and distribution is not considered.

In [8] a fault tolerance framework is presented which allocates tasks to fault tolerant computing nodes before the execution of tasks and also reassigns the tasks to the computing node with lower response time. In comparison with Bframework, its task model considers task as graph but the 3Vs model of streaming Big Data is not considered completely. Its resource model is similar to Bframework and model the capacity of the resources. Its scheduling model aims to minimize the response time and maximize the system reliability where in comparison with Bframework the workload distribution and balancing is not considered.

In [15], authors predict the data characteristics (5Vs) of streaming Big Data and expressed in CoBA values which is used to determine the required computing nodes of the tasks. In comparison with Bframework, It just considers the streaming Big Data and there is no model for task and resources.

In [7], a stable scheduling is presented which allocates tasks to the stable computing nodes before the execution of tasks and reassigns the task to the computing node with lower response time. Similar to Bframework, it models tasks as graph which encompasses operators and their input and outputs but the variety dimension of Big Data is not considered. Similar to Bframework, It models resources with capacity. Its scheduling model aim to minimize the response time and maximize the stability of the computing nodes. However, the workload balancing and distribution is not considered.

In [16], data characteristics of streaming Big Data based on the 4Vs model is estimated as a value named Characteristics of Data (CoD). Furthermore Self-Organizing Maps (SOM) are used to allocate cloud resources to streaming Big Data using its estimated CoD. In comparison with Bframework, this work only considers the streaming Big Data, the streaming Big Data queries are not considered.

In [17], authors derived effective optimization rules which are used to reconfigure the structure of an accepted task at runtime using real-time performance statistics. In comparison with Bframework, the result of this work can be used as an optimized input query to Bframework.

In summary, the current optimization strategies for improving the scheduling algorithm of streaming Big Data only consider some aspects of the Big Data characteristics. However, our proposed framework proposes scheduling algorithms which not only considers all of the mentioned aspects but it is also suited for the analytical queries of streaming Big Data.

3. Problem statement

In the context of stream Big Data computing, a set of streaming Big Data analytic queries is accepted by a stream Big Data computing platform and one or multiple streams of Big Data flow through the accepted queries. The platform’s scheduler handles the efficient assignment and scheduling of the query operators over the underlying cluster resources. The definitions below incorporate these factors into the scheduling decisions.

3.1 Big Data stream

In Bframework, a stream is a sequence of data sets $=\{{Ds}_{1},\ldots,{Ds}_{i},\ldots,{Ds}_{ds}\}$ , where each data set ${Ds}_{i}$ is a set of $t s$ number of $t_{ij}$ tuple ${Ds}_{i}=\{t_{i1},\ldots,t_{ij},\ldots,t_{its}\}$ . A continuous stream is an infinite sequence of data sets and parallel continuous streams are multiple streams that can be processed at the same time. Big data stream $B s$ is a continuous stream with high velocity, high volume and a wide variety of data sets. We applied the structured, semi-structured and unstructured definitions of Big Data [14] and formally define $B s$ using Definitions 1–4.

Definition 1 (Structured data stream). Structured data stream Sds is a continuous stream with the pre-defined schema Sche. Sche can be defined as a set of $f n$ fields $\textit{Sche}=\{f_{1},\ldots,f_{i},\ldots,f_{fn}\}$ , where each field $f_{i}$ is described by a double $(f_{i}^{\textit{name}},f_{i}^{\textit{size}})$ , where $f_{i}^{\textit{name}}$ is the name of $f_{i}$ and the $f_{i}^{\textit{size}}$ is the size of $f_{i}$ in byte. Furthermore, let the structured data flow Sdf be defined as an edge that flows Sds.

Definition 2 (Semi-structured data stream). Semi-Structured data stream Eds is a continuous stream having a special instance of schema Sche with $fn=2,$ where $f_{1}$ is defined as the key field and $f_{2}$ is defined as the value field. Furthermore, let the semi-structured data flow Edf be defined as an edge that flows Eds.

Definition 3 (Unstructured data stream). Unstructured data stream, Uds is a continuous stream with no schema defined. Furthermore, let the unstructured data flow Udf be defined as an edge that flows Uds.

Definition 4 (Big Data stream). The Big Data stream $B s$ is a continuous stream. The volume of its data sets is defined by ${Bs}^{Vo}=\{{Ds}_{1}^{Vo},\ldots,{Ds}_{i}^{Vo},\ldots,{Ds}_{ds}^{Vo}\}$ , velocity of its data sets is defined by ${Bs}^{Ve}=\left\{{Ds}_{1}^{Ve},\ldots,{Ds}_{i}^{Ve},\ldots,{Ds}_{ds}^{Ve}\right\}$ and the variety of its data sets is defined by ${Bs}^{Va}=\{{Ds}_{1}^{Va},\ldots,{Ds}_{i}^{Va},\ldots,{Ds}_{n}^{Va}\}$ . All data sets of a ${Bs}^{Va}$ can be members of $\{\textit{Sds},\textit{Eds},\textit{Uds}\}$ and a Big Data stream flow $B f$ can be member of $\{\textit{Sdf},\textit{Edf},\textit{Udf}\}$ .

3.2 Streaming Big Data analysis query

In Bframework, a Big Data stream analytic query can be defined as a directed graph with a set of operator instances which analyze one or multiple input Big Data streams $Bs(s)$ and produce one or multiple output Big Data streams $Bs(s)$ . We formally define the Big Data stream analytic query using Definition 5.

Definition 5 (Big Data stream analytic query). Big Data stream analytic query ${Bq}_{i}$ can be expressed as a directed graph ${Bq}_{i}=(VE)$ , where $V$ is a finite set of $v n$ operator instances $V=\{O_{1},\ldots,O_{i},\ldots,O_{vn}\}$ , and $E=\{E_{11},\ldots,E_{pq},\ldots,E_{\textit{vnvn}}\}$ is a finite set of directed edges, where $E_{pq}$ represents execution precedence between two operators $p$ and $q$ and is member of $B f$ .

The operator instance $O_{i}$ of $B f$ is characterized by a five-tuple $O_{i}=\left(O_{i}^{\textit{Inp}},O_{i}^{Id},O_{i}^{\textit{out}},O_{i}^{% \textit{Tpe}},O_{i}^{\textit{Ins}}\right)$ , where $O_{i}^{\textit{Inp}}$ is a set of input Big Data flows $Bf(s)$ to the operator $O_{i}$ . $O_{i}^{Id}$ is the identification of the operator $O_{i}$ and encompasses the instance number of $O_{i}$ , $O_{i}^{\textit{out}}$ is the set of output Big Data flow(s) $Bf(s)$ for the operator $O_{i}$ . $O_{i}^{\textit{Tpe}}$ indicates the type of $O_{ij}$ as SQL-like, user-defined, or data source operator and $O_{i}^{\textit{Ins}}$ is the instruction set of $O_{i}$ . The in-degree of operator $O_{i}$ is the number of incoming edges, and the out-degree of operator $O_{i}$ is the number of the outgoing edges. The source operator is the operator whose in-degree is zero, and the end operator is the operator whose out-degree is zero. A ${Bq}_{i}$ has at least one source and one end operators. In addition, we define two ${Bq}_{i}$ related concepts ${Bq}_{i}$ layer and ${Bq}_{i}$ segment. The layer of ${Bq}_{i}$ is a set of ${Bq}_{i}$ operators whose instance numbers are same and segment of ${Bq}_{i}$ is a set of ${Bq}_{i}$ operators whose source operators have same graph hop count.

The directed edge $E_{pq}$ is the directed communication link from the operator $O_{i}$ to $O_{j}$ and is characterized by a triple $E_{pq}=(E_{pq}^{\textit{Ide}},E_{pq}^{\textit{Tpe}},E_{pq}^{\textit{Sze}})$ , where $E_{pq}^{\textit{Ide}}$ is the identification of $E_{pq}$ , $E_{pq}^{\textit{Tpe}}$ is $B f$ type of $E_{pq}$ and $E_{pq}^{\textit{Sze}}$ is the $B f$ size of $E_{pq}$ .

An example of a ${Bq}_{i}$ with twenty-four vertices is shown in Fig. 1, where the vertices set is parallelized into two layers and the source vertices in the first layer are $v_{s_{{11}}}$ and $v_{s_{{12}}}$ and the end vertex is $v_{c}$ , the in-degree of join vertex ${{\triangleright\triangleleft}}_{11}$ is two, and the out-degree of join vertex ${{\triangleright\triangleleft}}_{11}$ is one.

Figure 1.

${Bq}_{i}$ a sample streaming Big Data analytic query.

3.3 Scheduling model of Big Data stream analytic query

In Bframework, the scheduler accepts a set of independent Big Data stream analytic queries, places, and allocates all operators of the accepted queries to the underlying cluster resources to achieve the lowest queries latency and to meet the cluster resource capability constraints. The following model reflects these factors.

Let $B Q$ denote a set of independent $b q$ queries accepted by the scheduler $BQ=\{{Bq}_{1},\ldots,{Bq}_{i},\ldots,\linebreak{Bq}_{bq}\}$ , where each query ${Bq}_{i}$ was defined by Definition 5. Furthermore, let $C N$ denote a set of $c n$ computing nodes $CN=\left\{{Cn}_{1},\ldots,{Cn}_{i},\ldots,{Cn}_{cn}\right\}$ , where each ${Cn}_{i}$ is described by a triplet ${Cn}_{i}=\left({Cn}_{i}^{Cu},{Cn}_{i}^{\textit{Cpu}},{Cn}_{i}^{\textit{Bwt}}\right)$ . Here, ${Cn}_{i}^{\textit{Cpu}}$ represents the CPU capacity of ${Cn}_{i}$ , ${Cn}_{i}^{\textit{Bwt}}$ is the network interface card speed of ${Cn}_{i}$ , and ${Cn}_{i}^{cu}$ is a set of assigned computing units $CU=\{{Cu}_{1},\ldots,{Cu}_{i},\ldots,{Cu}_{cu}\}$ to ${Cn}_{i}$ . A computing unit ${Cu}_{i}$ denotes a thread or process determined by Bframework where the thread computing unit is a set of the ${Bq}_{i}$ operators and the process computing unit is a set of threads identified by Bframework.

The schedule matrix ${\textit{Schedule}}_{(bq\times vn)\times cn}$ represents the assignment of all operators of all accepted queries to the computing nodes ${\textit{Schedule}}_{(bq\times vn)\times cn}=\left({\begin{array}[]{*{20}c}S_{% 11}&S_{bq\times vn1}\\ S_{1cn}&S_{bq\times vn\times cn}\\ \end{array}}\right)$ , where $S_{ij}$ shows that $O_{i}$ is assigned to the computing node ${Cn}_{j}$ . To keep the computing node ${Cn}_{i}$ in the operative ${Cn}_{i}$ we should meet the constraints defined by Eq. (1), so that assigned operators $S_{ij}.O_{i}^{\textit{Ins}}$ to ${Cn}_{j}^{\textit{Cpu}}$ would not exceed the rate at which operators can be processed.

$\displaystyle\sum\limits_{i=1}^{bq\times vn}S_{ij}.O_{i}^{\textit{Ins}}% \leqslant{Cn}_{j}^{\textit{Cpu}}$ (1)

In Bframework, the schedule model used by its scheduler is defined by Definition 6.

Definition 6 (Scheduling model $S m$ ). $S m$ is the scheduling model used in Bframework and is represented by four-tuple $Sm=\{BQ,CN,CO,OB\}$ , where $BQ=\{{Bq}_{1},\ldots,{Bq}_{i},\ldots,{Bq}_{bq}\}$ is the set of $b q$ accepted queries and $C N$ is the set of computing nodes as the underlying cluster resources of Bframework. $C O$ is the constraint defined by Eq. (1) and $O B$ is the objective function of $S m$ which is defined to minimize the average tuple latency of each ${Bq}_{i}$ of the accepted queries $B Q$ according to Eq. (2)

$\displaystyle OB\left({Bq}_{i}\right)=\min\textit{imize }{\textit{Avg}\left(% \textit{Latency}\left({Bq}_{i}\right)\right)},$ (2)

where $\textit{Latency}\left({Bq}_{i}\right)$ represents the tuple latency of ${Bq}_{i}$ and it is defined according to Eq. (3)

$\displaystyle\textit{Latency}\left({Bq}_{i}\right)=\left(\textit{Time}\left({% Bq}_{i}.O_{e}^{\textit{Out}}\right)-\textit{Time}\left({Bq}_{i}.O_{s}^{\textit% {Inp}}\right)\right),$ (3)

where $\textit{Time}\left({Bq}_{i}.O_{e}^{\textit{Out}}\right)$ is the timestamp of the earliest $B s$ of the output $B f$ in ${Bq}_{i}^{\prime}s$ end operator and $\textit{Time}\left({Bq}_{i}.O_{s}^{\textit{Inp}}\right)$ is the timestamp of the earliest $B s$ of the ${Bq}_{i}^{\prime}s$ source operators.

4. Bframework

The goal of Bframework is to minimize the average tuple latency of the queries which are accepted by Bframework. To achieve this goal, the idea is to use more than one scheduling strategy for ${Bq}_{i}$ operators. The rationale of this idea is related to an attribute of analytical queries where the output size of analytic query end operators are usually smaller than input size of its source operators [3]. Bframework uses this attribute to realize its idea by profiling the output size of ${Bq}_{i}$ operators and partitioning the operators using the profiled output size.

To support the idea, the objectives of Bframework assignment and scheduling algorithms are established based on the characteristics of the identified partitions. The objective of Bframework operator assignment, Off-line Bscheduler and Online Bscheduler algorithms are to minimize the inter operator traffic loads of the first partition operators and distribute and balance the workloads of the second partition operators. These algorithms applies their own assignment and scheduling strategies to achieve the objective of each partition.

To implement the needed strategies of each partition, the architecture of Bframework is designed which encompasses five modules and operates within an open source stream Big Data computing platform at two phases. Details of the Bframework architecture, phases and algorithms are explained in the following sections.

4.1 Bframework architecture

The architecture of Bframework includes five modules and the workflow of modules operates at two phases. In the first phase, prior to the execution of its accepted queries, the profiler and partitioner, the operator assigner and the off-line scheduler modules operate. In the second phase, during the execution of queries, the on-line rescheduler and the monitor modules work periodically. The overall view of the architecture and its workflow are presented in Fig. 2.

Figure 2.

Architecture of Bframework.

Prior to the execution of each query ${Bq}_{i}$ , the profiler and partitioner module accepts ${Bq}_{i}$ and examines its structure to estimate the value of a $B f$ metric named outBfsize, which indicates the estimated size of output $B f$ of an operator. Furthermore, this module partitions ${Bq}_{i}$ into at most three distinct partitions using the outBfsize value which are annotated in $Bq^{\prime}_{i}$ . For each partition, the operator assigner module first determines the initial number of thread computing units $C u$ which will be used for operator partitioning and then distributes the operators into threads using the outBfsize and another defined metric named simple selectivity metric, which is explained in Section 3.2.2. The off-line scheduler module determines the initial number of process computing units and balances the workload of the determined threads to the identified processes. Furthermore, it also assigns the identified processes to the available computing nodes based on the resource capability of the computing nodes. The output of the last module is a schedule Schedule containing the identified computing nodes and their assigned operators.

During the execution of each query ${Bq}_{i}$ at the second phase, the monitor module continuously monitors the traffic loads and workloads of the threads and especially some operators of the ${Bq}_{i}$ and also the workload of the available computing nodes. Furthermore, this module periodically triggers the online scheduler module with the monitored loads of threads, operators and computing nodes. The online re-scheduler module periodically reacts to the overloaded computing nodes and the current workload of the second partition to find a more efficient scheduler $\textit{Schedule}^{\prime}$ . This module also reacts to the traffic load of the operators which flow $B f$ between the layers of the ${Bq}_{i}$ . In addition, this module make decisions about the assignment and scheduling of the third partition operators in its first execution iteration.

4.2 Bframework algorithms

This section presents the detail of the algorithms used in Bframework. First, it is describes the query profiling and partitioning algorithm, which is used to partition the query ${Bq}_{i}$ . Then the operator assignment algorithm is described, which is applied to assign the operators of the partitioned ${Bq}_{i}$ to the computing units. Finally, it describes two scheduling algorithms Off-line Bscheduler and Online Bscheduler, used for scheduling ${Bq}_{i}$ operators to the computing nodes before and during the execution of ${Bq}_{i}$ respectively.

4.2.1 Query profiling and partitioning algorithm

The query profiling and partitioning algorithm profiles and partitions the operators of the query ${Bq}_{i}$ prior to the execution of ${Bq}_{i}$ . The objective is to partition the operators of ${Bq}_{i}$ in way that the different scheduling strategies can be applied for each partition. This algorithm uses the outBfsize metric to partition ${Bq}_{i}$ into at most three distinct operator partitions named ${Bq}_{i}^{p1},{Bq}_{i}^{p2},{Bq}_{i}^{p3}$ . As a result, the partition ${Bq}_{i}^{p1}$ has the operators whose average size of the outBfsize is greater than that of the outBfsize for the operators of partition ${Bq}_{i}^{p2}$ . The partition ${Bq}_{i}^{p3}$ includes the remaining operators whose values of outBfsize could not be estimated at the first phase. The aim of profiling is to estimate the value of outBfsize metric and determine the segments of ${Bq}_{i}$ . The value of outBfsize metric is estimated according to Eq. (4).

$\displaystyle\textit{outBfsize}\left(O_{i}\right)=\left\{{\begin{array}[]{ll}% \sum\limits_{{j=0}}^{\textit{out-degree}\left(O_{i}\right)}\sum\limits_{{k=0}}% ^{k=f_{n}}O_{i}^{\textit{out}_{j}}.f_{k}^{\textit{size}}&\text{if }\textit{% phase}=1\text{ AND }O_{i}^{\textit{out}}\in\left\{\text{Sdf,Edf}\right\}\\ \infty&\text{if }\textit{phase}=1\text{ AND }O_{i}^{\textit{out}}\in\left\{% \text{Udf}\right\}\\ \sum\limits_{{j=0}}^{\textit{out-degree}\left(O_{i}\right)}\sum\limits_{{k=0}}% ^{k=ds}O_{i}^{{\textit{out}}_{j}}.Ds_{k}^{Vo}&\text{if }\textit{phase}=2\\ \end{array}}\right.$ (4)

In Eq. (4) $\sum_{{k=0}}^{k=f_{n}}O_{i}^{\textit{out}_{j}}.f_{k}^{\textit{size}}$ indicates the sum of the size of the $f n$ fields of the $j$ -th output $B f$ of $O_{i}$ and $\sum_{k=0}^{k=ds}O_{i}^{\textit{out}_{j}}.Ds_{k}^{Vo}$ denotes the sum of the size of the recent data sets, based on the window size, of the $j$ -th output $B f$ of $O_{i}$ . In the first phase, outBfsize value is determined based on the type of the output $B f$ . In Sdf or Edf cases, the size of the field is used to calculate the value of the outBfsize. Nevertheless, in Udf case, outBfsize value cannot estimate at the first phase. Algorithm 1. describes how this metric is used to partition the ${Bq}_{i}$ operators.

Algorithm 1:

{Bq}_{i}

profiling and partitioning

Input:

{Bq}_{i}

: query

Output:

{Bq^{\prime}}_{i}^{\text{p1}}

{Bq^{\prime}}_{i}^{\text{p2}}

and

{Bq^{\prime}}_{i}^{\text{p3}}

: profiled partitions

\textit{sourceoperators}\leftarrow{Bq}_{i}.\textit{sources}

\textit{endoperator}\leftarrow{Bq}_{i}.\textit{end}

\textit{sdf}\leftarrow\text{Avg}(\text{OutBfsize}(\textit{sourceoperators}^{% \text{Inp}}))

\textit{ddf}\leftarrow\text{OutBfsize}(\textit{endoperator}^{\text{Out}})

\textit{firstlayer}\leftarrow\text{Layer}({Bq}_{i},1)

\textit{segment}=1

7. for

(\textit{operator}:\text{TopologicalOrder}(\textit{firstlayer}.V))

8. if

(|\textit{operator}^{\text{Inp}}|>1\quad\textbf{AND }\text{Parents}(\textit{% operator})\in\textit{sourceoperators})

{Bq}_{i}.\textit{operator}^{\text{sg}}=\textit{segment}

10.

\textit{segment}++

11. end if

12.

\textit{sdistance}=|\text{OutBfsize}(\textit{operator}^{\text{Inp}})-\textit{% sdf}|

13.

\textit{ddistance}=|\text{OutBfsize}(\textit{operator}^{\text{Inp}})-\textit{% ddf}|

14. if

(\text{OutBfsize }(\textit{operator}^{\text{Inp}})==\infty){Bq}_{i}.\textit{% operator}^{\text{p}}=

‘

p3

’

15. if

(\textit{sdistance}\leqslant\textit{ddistance}){Bq}_{i}.\textit{operator}^{% \text{p}}=

‘

p1

’

16. if

(\textit{sdistance}>\textit{ddistance}){Bq}_{i}.\textit{operator}^{\text{p}}=

‘

p2

’

17.

{Bq}_{i}.\textit{operator}^{\text{sg}}=\textit{segment}

18.

{Bq}_{i}.\textit{operator}^{\text{se}}=\textit{Selectivity}(\textit{operator})

19. end for

20.

{Bq}_{i}.\textit{segments}=\textit{segment}

21. for

(\textit{operator}:\text{TopologicalOrder}({Bq}_{i}.V-\textit{firstlayer}.V))

22.

\text{Annotate}(\textit{operator},\text{Layer}(\textit{operator},1))

23. end for

24.

\textit{return }{Bq^{\prime}}_{i}^{p1},{Bq^{\prime}}_{i}^{p2},{Bq^{\prime}}_{i% }^{p3}

The algorithm operates in three main steps:

Lines 1–4: Calculation of the outBfsize metric of the source and end operators located in the first layer of ${Bq}_{i}$ . As there may be more than one source operator, the average value of outBfsize is calculated.

Lines 5–20: Calculation of the outBfsize value of all ${Bq}_{i}$ operators located in the first layer to determine the corresponding partition of each operator. The algorithm computes the distance of the outBfsize of the operator with both average outBfsize of the source operators and outBfsize of the end operator. The partition ${Bq}_{i}^{p1}$ includes the operators of ${Bq}_{i}$ which has the minimum distance with the source operators. Whereas, the partition ${Bq}_{i}^{p2}$ encompasses the ${Bq}_{i}$ operators which has the minimum distance with the end operator. In case that the value of the outBfsize for the operator is not estimable at the first phase, the operator is added to the partition ${Bq}_{i}^{p3}$ .

Lines 21–24: Annotation of the corresponding partition of all operators in the remaining layers. The algorithm sets the partition attribute of an operator with its corresponding operator in the first layer.

In order to profile the segments of ${Bq}_{i}$ , the algorithm identifies the operators whose input $B s$ streams are more than one input $B s$ and whose parent operators are source operators. The identified operators and all its successor operators placed in the new segment except that the mentioned condition happens again in the remaining operators of the first layer of ${Bq}_{i}$ . Figure 1 shows an example of the identified segments and its corresponding operators.

4.2.2 Operator assignment algorithm

The operator assignment algorithm aims to efficiently assign the operator of each identified partition to a set of initial thread computing units and also assigns the identified thread to a set of initial process computing units prior to the execution of the operators. This algorithm uses three different assignment strategies to provide an efficient operator assignment for each partition.

The assignment strategy for the first partition is to minimize the inter-operator traffic load through minimizing the inter-thread and the inter-process traffic loads, because the inter-operator traffic load of the ${Bq}_{i}^{p1}$ operators is the most time-consuming factor for the tuple latency of the query ${Bq}_{i}$ . Therefore, the algorithm assigns the most operators of ${Bq}_{i}^{p1}$ to at least thread computing units and also assigns the most identified thread to at least process computing units.

The assignment strategy for the second partition is to distribute and balance the workload of the ${Bq}_{i}^{p2}$ operators through distributing and balancing the workload of ${Bq}_{i}$ thread and process computing units. As result, the average waiting time of the accepted queries could be decreased. To realize the strategy, the algorithm first determines the needed number of thread and process computing units for this partition and then distributes the ${Bq}_{i}^{p2}$ workloads into the identified threads using a defined distribution criteria named simple selectivity. Finally, this algorithm balances the threads workloads of the identified processes through the simple defined balancing method. The first step of the algorithm is realized using the simple selectivity criteria which has simply four distinct values addressed in Table 1. From Table 1. $m$ denotes the number of input tuples, $n$ indicates the number of the output tuples and $\infty$ notes the simple selectivity value of an operator cannot estimate at the first phase. By using simple selectivity metric as distribution criteria, it is possible to distribute operators with similar runtime output size into the same thread computing units. For example, two ${Bq}_{i}^{p2}$ operators which are connected directly through an $B f$ , placed in the same partitions and have same simple selectivity values are assigned to a thread computing unit and will be rescheduled together if Online Bscheduler decides to reassign them. To balance the workload of the identified threads to process, a simple balancing method is used in which the thread whose workload is largest is selected first and then the descending order of the remaining thread workloads is examined to orderly group threads whose workload size is near to the workload size of the selected thread.

Table 1
Simple selectivity criteria

Operator type	Operator input/output rate	Simple selectivity	Example
SQL	$m/n=1$	1	Project
SQL	$m/n\leqslant 1$	$<$ 1	Selection
SQL	$m/n>1$	$>$ 1	Join
User-defined	–	$\infty$	–

The assignment strategy for the third partition is to simply use a thread and a process computing units and assigns all the ${Bq}_{i}^{p3}$ operators to the identified thread and also assigns the thread to the process, because it is impossible for this algorithm to estimate the value of outBfsize for the third partition at the first phase. The implementation details of these strategies are presented in Algorithm 2.

Algorithm 2:

Bq_{i}

operator assignment

Input:

{Bq^{\prime}}_{i}^{\text{p1}}

{Bq^{\prime}}_{i}^{\text{p2}}

and

{Bq^{\prime}}_{i}^{\text{p3}}

: profiled partitions

Output:

\text{CU}^{\text{thread}}

\text{CU}^{\text{process}}

: thread and process computing units

p1

p2

p3

\leftarrow

1 // iteration for the partitions

2. for (

s:Bq_{i}.\textit{segments}

)

3. for (

\textit{operator}:Bq_{i}^{(\text{sg}=\text{s})}.\textit{operators}

)

4. switch (

\textit{operator}^{\text{p}}

)

5. case ‘

p1

’:

6. if (|

\textit{operator}^{\text{Inp}}

>

op=\textit{Max}(\textit{OutBfsize}(\textit{operator}^{\text{Inp}}))

Bq_{i}^{\text{p1}}.\textit{operators}-=\{op,\textit{operator}\}

Cu_{(\text{p1}++)}^{\text{thread}}=\{op,\textit{operator}\}

10. break

11. end if

12. if (

\textit{operator}^{\text{Tpe}}\in\{\textit{Source},\textit{User}-\textit{% Defined}\}

)

13.

Bq_{i}^{\text{p1}}.\textit{operators}-=\{\textit{operator}\}

14.

Cu_{(\text{p1}++)}^{\text{thread}}.\textit{operators}=\{\textit{operator}\}

15. break

16. end if

17.

Cu_{(\text{p1}++)}^{\text{thread}}.\textit{operators}=Bq_{i}^{\text{p1}}.% \textit{operators}

18. case ‘

p2

’:

19. if (

\textit{operator}^{\text{Tpe}}\in\{\textit{User}-\textit{Defined}\})

20.

Bq_{i}^{\text{p2}}.\textit{operators}-=\{\textit{operator}\}

21.

Cu_{(\text{p2}++)}^{\text{thread}}.\textit{operators}=\{\textit{operator}\}

22. break

23. end if

24. while (

\textit{operator}^{\text{se}}=\textit{Next}(\textit{operator}^{\text{se}})

)

25.

Cu_{\text{p2}}^{\text{thread}}.\textit{operators}=\textit{operator}\bigcup% \textit{Next}(\textit{operator})

26.

\textit{operator}=\textit{Next}(\textit{operator})

27. end while

28.

Cu_{(\text{p2}++)}^{\text{thread}}=\textit{operator}

29. case ‘

p3

’:

30.

Cu_{\text{p3}}^{\text{thread}}.\textit{operators}=\textit{operator}

31. end switch

32. end for

33. end for

34.

CU^{\text{thread}}\leftarrow\text{Sort}(CU^{(\text{thread}(\text{p}=\text{p2})% )},\textit{workload},\textit{Desc})

35.

wl\leftarrow\text{First}(CU^{\text{thread}})

36.

p1

p2

p3\leftarrow 1

37. for (

Cu:CU^{\text{thread}}

)

38. switch (

Cu^{\text{p}}

)

39. case ‘

p1

’:

40. if (

Cu.\textit{operators}\in\{\textit{User}-\textit{defined}\}

)

41.

CU^{\text{thread}}-=Cu

42.

Cu_{(\text{p1}++)}^{\text{process}}.\textit{threads}=Cu

43. break

44. end If

45.

Cu_{(\text{p1}++)}^{\text{process}}.\textit{threads}=Cu

46. break

47. case ‘

p2

’:

48. if (

Cu.\textit{operators}\in\{\textit{User}-\textit{defined}\}

)

49.

CU^{\text{thread}}-=Cu

50.

Cu_{(\text{p2}++)}^{\text{process}}.\textit{threads}=Cu

51. end If

52. while (

Cu.\textit{workload}\leqslant wl

)

53.

CU^{\text{thread}}-=Cu

54.

Cu_{(\text{p2}++)}^{\text{process}}=Cu

55. end while

56.

Cu_{(\text{p2}++)}^{\text{process}}.\textit{threads}=Cu

57. break

58. case ‘

p3

’:

59.

Cu_{\text{p3}}^{\text{process}}.\textit{threads}=Cu

60. end switch

61. end for

To implement the assignment strategy for the first partition, the algorithm initially assigns all operators of ${Bq}_{i}^{p1}$ to a thread computing unit. Two conditions cause the assigned operators to be excluded from the used thread and which are then assigned to a new thread. As first condition, if the assigned operator is type of the source or user-defined operator, these operators will be excluded which leads to two effects. First, for the source operators, Bframework will be flexible to schedule the source operator on the data source nodes that can result in the inter-node traffic reduction. And second, for the user-defined operators, Bframework will be able to schedule these operators independence of their predecessors or successors based on the determined workload at the second phase. As second condition, if the number of input $B s$ streams of an operator is greater than one, the assigned operator and one of its parent operators will be excluded and assigned to a new thread. For parent selection, the parent operator whose output $B s$ for the assigned operator is greatest will be selected to exclude from the thread computing unit. By this decision, the inter-thread traffic load may be decreased (Lines 5–17). For the process assignment, the algorithm assigns the most threads of the first partition to at least the processes, so the algorithm initially assigns all identified threads of each segment to a process computing unit (Lines 39–46).

To implement the assignment strategy for the second partition, the algorithm traverses the ${Bq}_{i}^{p2}$ operators and examines the simple selectivity value of the operator operator and its successor. If these operators have same simple selectivity values this algorithm will assign them to a thread (Lines 18–28). To balance workload of the identified threads, firstly, all identified threads are sorted in descending order by their workload and the first thread is selected as maximum workload thread. Secondly, all remaining threads are assigned orderly to a process computing unit until process workload reaches the selected workload thread (Lines 47–57).

To implement the assignment strategy for the third partition, all the operators of this partition are simply assigned to a thread (Lines 29–30) and process computing units (Lines 58–59).

4.2.3 Off-line Bscheduler

The off-line Bscheduler aims to assign the identified processes to a set of available computing nodes prior to the execution of a query ${Bq}_{i}$ . The algorithm uses three different assignment strategies for each partition. The Algorithm 3 presents the details of these strategies.

Algorithm 3:

{Bq}_{i}

off-line scheduling

Input:

{\text{CU}}^{\text{process}}

: process computing unit

Output:

\textit{Schedule}:\text{Schedule }

CN\leftarrow\textit{Sort}(CN,{Cn}^{\text{Cpu}},\textit{Desc})

Cn\leftarrow\textit{First}(CN)

3. for

({Cu}_{i}^{\text{process}}:{CU}^{\text{process}(\text{p}=\text{p1})})

4. if (meet(constraint Eq. (1)))

\textit{Schedule}[Cn,{Cu}_{i}^{\text{process}}.\textit{operators}]\leftarrow 1

Cn.\textit{Assigned}=\textit{True}

7. else

Cn\leftarrow\textit{Next}(CN)

9. end if

10. end for

11.

CN-={CN}^{\text{Assigned}=\text{true}}

12. for

({Cu}_{i}^{\text{process}}:{CU}^{\text{process}(\text{p}=\text{p2})})

13. if (meet(constraint Eq. (1)))

14.

\textit{Schedule}\left[Cn,{Cu}_{i}^{\text{process}}.\textit{operators}\right]\leftarrow 1

15.

Cn.\textit{Assigned}=\textit{True}

16.

Cn\leftarrow\textit{Next}(CN)

17. else

18.

Cn\leftarrow\textit{Next}(CN)

19. end if

20. end for

21.

CN-={CN}^{\text{Assigned}=\text{True}}

22. for

({Cu}_{i}^{\text{process}}:{Cu}^{\text{process}(\text{p}=\text{p3})})

23. if (meet(constraint Eq. (1)))

24.

\textit{Schedule}[Cn,{Cu}_{i}^{\text{process}}.\textit{operators}]\leftarrow 1

25. else

26.

Cn\leftarrow\textit{Next}(CN)

27. end if

28. end for

For the first partition, the objectives of this algorithm is to reduce the inter-node traffic load of the ${Bq}_{i}^{p1}$ operators and improve the processing speed of ${Bq}_{i}^{p1}$ operators respectively. As traffic load is dominate time consuming factor of ${Bq}_{i}^{p1}$ , the first objective achievement can lead to the tuple latency reduction. By achieving the second objective, all the ${Bq}_{i}$ remaining operators can start their processing earlier. Therefore, the scheduling strategy of this partition is to assign the most operators of the ${Bq}_{i}^{p1}$ to at least computing nodes whose CPU capacity are highest. To implement this strategy (Lines 1–10), the algorithm sorts the available computing nodes in descending order by their CPU capacity and assigns at least available computing nodes to the most processes of this partition with respect to the constraint Eq. (1).

For the second partition, the algorithm aim to balance the workload of ${Bq}_{i}^{p2}$ to reduce the average waiting time of the accepted queries. Therefore, the scheduling strategy is to simply assign each process of this partition to the ordered unassigned computing nodes. This algorithm reserves the ${Bq}_{i}^{p1}$ assigned computing nodes for assigning them to the first partition of other accepted queries (Lines 11–20). This decision increases the availability of the highest CPU capacity computing nodes for other first partitions. In addition, using the ordered computing nodes can improve the processing speed of the ${Bq}_{i}^{p2}$ operators.

For the third partition, the scheduling strategy is to assign all ${Bq}_{i}^{p3}$ processes to at least remaining computing nodes. This strategy can minimize the possible inter-node traffic load of ${Bq}_{i}^{p3}$ operators (Lines 21–28).

4.2.4 Online Bscheduler

The online Bscheduler aims to reduce the ${Bq}_{i}$ tuple latency through the periodic efficient reassignment of current threads and some operators of ${Bq}_{i}$ to the new computing nodes in reaction to the $B s$ fluctuations during the ${Bq}_{i}$ execution. Furthermore, the algorithm efficiently assigns the operators of the third partition to a set of processes and to a set of computing nodes. This algorithm also reacts to the overloaded computing nodes. The details of the algorithm are presented in Algorithm 4.

Algorithm 4:

{Bq}_{i}

online scheduling and rescheduling

Input:

{CU}^{\textit{thread}}

: thread computing unit, OverCn: overloaded computing node, operator:

{Bq}_{i}

operator

Output: Schedule’: Schedule

p1\leftarrow\text{average}(\textit{OutBfsize}({CU}^{\text{thread}(\text{p}=% \text{p1})}))

p2\leftarrow\text{average}(\textit{OutBfsize}({CU}^{\text{thread}(\text{p}=% \text{p2})}))

\textbf{for}(\textit{thread}:{CU}^{\text{thread}(\text{p}=\text{p3})})

\textbf{if}(\textit{OutBfsize}(\textit{thread})-p1>\textit{OutBfsize}(\textit{% thread})-p2)

{Bq}_{i}^{\text{p1}}=\textit{thread.operators}

6. else

{Bq}_{i}^{\text{p2}}=\textit{thread.operators}

8. end if

\textit{process}=\text{Operator Assignment Algorithm}({Bq}_{i})

10.

\textit{node}=\text{Offline Bscheduler}(\textit{process})

11.

\textit{Schedule}[\textit{node},\textit{process.operatos}]\leftarrow 1

12. end for

13.

\textbf{if}(\textit{OverCn}\in{CN}^{\text{p}=\text{p1}})

14.

{CU}^{\text{thread}}=\text{Sort}({CU}^{\text{thread }\left(\text{cn=OverCn}% \right)},\text{TrafficLoad, Asc})

15.

CN=\text{Sort }\left({CN}^{\text{p}=\text{p1}},\text{Workload},\text{Asc}\right)

16. else

17.

{CU}^{\text{thread}}=\text{Sort}({CU}^{\text{thread }\left(\text{cn=OverCn}% \right)}\text{Workload, Desc)}

18.

{CN}^{\text{p}=\text{p2}}=\text{Sort }\left({CN}^{\text{p}=\text{p2}},\text{% Workload},\text{Asc}\right)

19. end if

20.

\textbf{while}\left(\textit{OverCn.Load}>\textit{Threshold}\right)

21.

{Cu}^{\text{thread}}=\text{Next}({CU}^{\text{thread}})

22.

\textit{process}=\text{Neighbor}({Cu}^{\text{thread}})

23.

\textit{node}=\text{Next}(CN)

24.

\textit{Schedule}^{\prime}\left[\textit{node},\textit{process.operators}\right% ]\leftarrow 1

25. end while

26.

\textbf{for}\left(Cu:{Cu}^{\text{thread}(\text{p}=\text{p2})}\right)

27.

\textit{load}=\text{Min}\left(\left\{CN-{Cu}^{\text{node}}\right\},\text{% workload}\right)

28.

\textbf{ if}\left(\textit{load}<{Cu}^{{\text{cn}}^{\text{laod}}}\right)

29.

\textit{process}=\text{Neighbor}(Cu)

30.

\textit{Schedule}^{\prime}\left[\text{Node}\left(\textit{load}\right),\textit{% process.operators}\right]\leftarrow 1

31. end if

32. end for

33.

{\textit{Edges}}_{pq}=\text{Sort}(\bigcup_{\forall p,q\textit{ where }\left|{% \textit{operator}}^{\textit{Inp}}\right|>\text{1 and p.layer <> q.layer}}E_{pq% },\text{TrafficloadDesc})

34.

\textbf{for}(\textit{Edge:Edges}_{pq})

35.

\textbf{if}(\text{Node}\left(\textit{Edge.p}\right)!=\text{Node}\left(\textit{% Edge.q}\right))

36.

\textit{Schedule}\left[\text{Node}(\textit{Edge.p}),\textit{Edge.q}\right]\leftarrow 1

37. end if

38. end for

To assign the operators of the third partition, the algorithm selects one of the two sets of the strategies used for the first or second partitions. To select the appropriate set of strategies, the algorithm compares the average of the outBfsize of each thread in the third partition with the average outBfsize of both first partition threads and second one. The algorithm selects the strategies of partition whose distance with the third partition is smallest (Lines 1–12).

To react to the overloaded computing nodes, two strategies are used by the algorithm. If the overloaded node hosts ${Bq}_{i}^{p1}$ operators, the algorithm selects at least to-be-reassigned threads whose traffic loads are smallest. Otherwise, the ${Bq}_{i}^{p1}$ inter-node traffic load can be increased by frequent threads reassignment. If the overloaded node hosts ${Bq}_{i}^{p2}$ operators, the algorithm selects the to-be-reassigned thread whose workload load is largest. It may cause the overloaded node exits its overloading state faster. After thread selection, the algorithm proceed to select the new related destination process. For both partitions, the algorithm choices the process whose threads has directed edge with the selected thread. To select the new destination node, node with the smallest workload is chosen to decrease the likelihood of a new overloading occurrence (Lines 13–25).

To react to the $B s$ fluctuations, this algorithm periodically evaluates the ${Bq}_{i}^{p2}$ thread workloads to find a more efficient new host node. To select the new host for a ${Bq}_{i}^{p2}$ thread, a host whose workload is smallest will be evaluated and selected if the new host workload is lower than the workload of current host of the thread. To select the new related destination process for the selected thread, the process whose threads has directed edge with the selected thread is chosen. This algorithm prefers to has no reaction to the ${Bq}_{i}^{p1}$ thread workloads to keep the inter node traffic load low (Lines 26–32).

Furthermore to react to $B s$ fluctuations, this algorithm aims to reduce the inter-layer traffic loads of the ${Bq}_{i}$ operators. This is because the inter-layer traffic load also affects the ${Bq}_{i}$ tuple latency. Therefore the algorithm periodically evaluates the ${Bq}_{i}$ operators whose input $Bf(s)$ are more than one $B f$ and at least one of its input $Bf(s)$ are not located in the same layer. Then, the algorithm assigns such operator on the host of its predecessor operator if the predecessor operator has the direct $B f$ with the operator and has the maximum $B f$ traffic load (Lines 33–38).

4.3 Bframework algorithms complexity analysis

The main cost of the profiling and partitioning algorithm comes from the sorting and traversing ${Bq}_{i}$ operators. Thus, the time complexity of the algorithm is $O(vn\log vn+vn)$ , where $v n$ represents the number of ${Bq}_{i}$ operators. In the operator assignment algorithm, the computation load is primarily made up of two parts. The operator assignment and thread computing unit assignment where for the operator assignment, the time expense of traversing operators is $O(vn)$ and for the thread assignment, the time cost of thread sorting and assignment is $O(|CU^{\textit{thread}}|\log{|CU}^{\textit{thread}}|+|CU^{\textit{thread}}|)$ . Therefore, the total time complexity of the algorithm is $O(vn+|CU^{\textit{thread}}|\log|CU^{\textit{thread}}|+|CU^{\textit{thread}}|)$ .

For Off-line Bscheduler, the main step is to assign the identified process computing units to the sorted available computing nodes. Thus, the time complexity of sorting computing nodes and the traversing processes computing units is $O(|CU^{\textit{process}}|+\textit{cnlogcn})$ . Online Bscheduler, in worst case, sorts and reassigns $v n$ operators, reassigns $|CU^{\textit{thread}}|$ threads and evaluates $c n$ computing nodes. Hence, as $vn\geqslant|CU^{\textit{thread}}|$ , the time complexity of this algorithm is $O(vn+vn\log vn+cn)$ .

5. Evaluation

This section describes the experiments that have been performed to validate and analyze the performance of Bframework. First, the query profiling and partitioning algorithm of Bframework is validated by comparing the identified partitions of the real-world and benchmark queries prior to their executions with the identified partitions after execution of the queries. Then, Bframework performance is assessed by analyzing the tuple latency of the benchmark and real-worlds queries in presence of different $B s$ fluctuations scenarios.

Table 2
Characteristics of BigBench queries [1]

Query type	Queries	Data types	Queries
Declarative	6, 7, 9, 13, 14, 16, 17, 19, 21, 22, 23, 24	Structured	1, 6, 7, 9, 13, 14, 15, 16, 17, 19, 20, 21, 22, 23, 24, 25, 26, 29
Mixed	1, 4, 5, 8, 11, 12, 15, 18, 20, 25, 26, 29, 30	Semi-structured	2, 3, 4, 5, 8, 12, 30
Procedural	2, 3, 10, 27, 28	Unstructured	10, 11, 18, 27, 28

5.1 Queries

The benchmark queries used in the experiments have been applied from the BigBench [14] and Linear Road Benchmark [1] benchmarks. BigBench includes the complex queries which cover the declarative and procedural and also the structured, semi-structured and unstructured aspects of the Big Data analytic queries. In other words, BigBench queries includes SQL-like and the user-defined operators and also, the structured, semi-structured and the unstructured data types. Table 2 describes the characteristics of the BigBench queries.

Linear Road Benchmark queries encompasses 17 simple stream queries whose operators are SQL-like and their data types are structured. Linear Road Benchmark has been used as benchmark for the evaluation of the distributed stream management systems. The WordCount and Yahoo page load processing are two real-world queries [19] which are usually used as test queries in the literature.

5.2 Environment

Table 3 describes the Storm cluster specifications used in the experiment to run the queries. The coordinating computing nodes #1 and #2 run zookeeper and nimbus and the worker computing nodes #3–15 execute the queries. The software stack used in the experiments is Ubuntu 14.04, Jdk 1.8, Apache Storm 0.9.5 [4] and MySQL 5.5.

Table 3
Storm cluster specifications

Computing node #	CPU	Memory	Bandwidth	Description
1	2 core–Intel G2020 2.9 GHz	1 GB	100 Mbps	Zookeeper
2	2 core–Intel G2020 2.9 GHz	1 GB	100 Mbps	Nimbus
3–6	Intel Core i5 3470 3.6 GHz	4 GB	100 Mbps	Workers
6–10	2 core–Intel G-2020 2.9 GHz	2 GB	100 Mbps	Workers
11–15	Intel Celeron D 3.2 GHz	2 GB	100 Mbps	Workers

5.3 Setup

All the benchmark and real-world queries are implemented using Trident API [22] of Apache Storm and Table 4 describes the parameters used in Bframework to execute the queries.

Table 4
Parameters used in Bframework

Parameter	Definition	Value
${Bs}^{Ve}$	The fixed average rate of $B f$	1000 tuples/s
	The fluctuated average rate of $B f$	[10000 tuples/s–30000 tuples/s]
$d s$	The number of Data sets (window)	30 seconds
${Bs}^{Va}$	The tuple size	[4B–400KB]
Threshold	The $C n$ overload threshold	70%
Execution	The execution period	15 minutes
Layer	The user defined layers of query	[1–2]

The range of values for ${Bs}^{Ve}$ is determined based on the conducted experiments to evaluate Bframework in face of the fixed low rate and high fluctuating rate of the incoming Big Data streams. Furthermore, the value of $d s$ is also determined by the experiment to reflect the behavior of the windows-based operators of the benchmark and real-world queries. The value of threshold parameter is also determined by the conducted experiments which is used by Bframework to react to the overloading condition of the computing nodes at runtime. The range of values for ${Bs}^{Va}$ is set using the structure of benchmark or real-world queries to evaluate Bframework algorithms in face of the different tuple sizes. In addition, range of values for the layer parameter is determined by the simple or complex type of the benchmark queries. Each query is run for 20 min and the average tuple latency of last 15 minutes, execution parameter, is estimated so the first 5 minutes initial stabilizing period is emitted from the experiment results. The tuple latency measurement of Strom is used for all the experiments.

5.4 Bframework validation

In order to validate Bframework, we validate the functionality of the query profiling and partitioning algorithm under the simple and complex benchmark queries and also under the real-world queries. The validation is conducted through comparing the identified partitions prior to execution of the queries with the identified partitions after their execution. In order to compare the identified partitions, the ratio between the average outBfsize of the first partition and the average outBfsize of the second one is calculated and shown in the R column of the result tables. Furthermore the number of identified partitions at the first phase is compared with the second one and is reported in the P column of the following tables.

Table 5
Result of partitioning Linear Road Benchmark queries at first phase

Query	R	P	Query	R	P	Query	R	P	Query	R	P
Q1	0%	2	Q6	48.5%	2	Q11	27.3%	2	Q16	0%	2
Q2	50%	2	Q7	0%	2	Q12	36.8%	2	Q17	0%	2
Q3	52.9%	2	Q8	20%	2	Q13	55.6%	2
Q4	0%	1	Q9	42.5%	2	Q14	13%	2
Q5	4.8%	2	Q10	27.3%	2	Q15	13%	2

R: outBfsize rate between first and second partitions; P: Number of identified partitions.

Table 6

Result of partitioning Linear Road Benchmark queries at second phase

Query	R	P	Query	R	P	Query	R	P	Query	R	P
Q1	1.8%	2	Q6	49.8%	2	Q11	30.9%	2	Q16	10%	2
Q2	51.3%	2	Q7	27.5%	2	Q12	45.3%	2	Q17	18%	2
Q3	56.5%	2	Q8	26%	2	Q13	57.8%	2
Q4	3.9%	1	Q9	49.4%	2	Q14	18.3%	2
Q5	4.9%	2	Q10	40%	2	Q15	27%	2

R: outBfsize rate between first and second partitions; P: Number of identified partitions.

5.4.1 Simple queries

Tables 5 and 6 show the result of partitioning Linear Road Benchmark queries before and after execution of the queries respectively. To execute these experiments, the configuration parameters ${Bs}^{Ve}$ and ${Bs}^{Va}$ are set to [10000 tuples/s–30000 tuples/s] and [4B–45B] respectively, which are derived from the structure of the Linear Road Benchmark queries.

The Table 5 demonstrates that there is no third partition prior to execution of Linear Road Benchmark queries. This is because all $B f$ of the queries are Sdf, and thus two partitions ${Bq}_{i}^{p1}$ and ${Bq}_{i}^{p2}$ are identified by the algorithm. Furthermore, the table also shows that the average rate of most identified partitions is low which is related to the tuple size of the queries. Since the distance of the tuple size of the first partition operators from the second one is small, the rate remains low. In addition, the table notes that, for the query Q4, the number of identified partitions is identified as 1 which is affected by the simple structure of the query. Table 6 shows that for most Linear Road Benchmark queries, in spite of the $B s$ fluctuation, the number and type of the identified partitions at the second phase is still identical to the identified partitions at the first phase. Meanwhile, for queries like Q7, the rate of the query in the second phase is greater than the first phase because the outBfsize of the identified partitions are the same.

Table 7
Result of partitioning BigBench queries at first phase

Query	R	P	Query	R	P	Query	R	P	Query	R	P	Query	R	P	Query	R	P
Q1	96.7%	2	Q6	80.2%	2	Q11	89.2%	3	Q16	94.2%	2	Q21	37.1%	2	Q26	86.2%	2
Q2	59.0 %	2	Q7	97.8%	2	Q12	83.0%	2	Q17	99.0%	2	Q22	87.4%	2	Q27	$\infty$	1
Q3	93.2%	2	Q8	71.7%	2	Q13	57.8%	2	Q18	74.5%	3	Q23	80.4%	2	Q28	$\infty$	1
Q4	39.9%	2	Q9	98.1%	2	Q14	97.2%	2	Q19	38.5%	2	Q24	89%	2	Q29	93.8%	2
Q5	86.8%	2	Q10	$\infty$	1	Q15	95.2%	2	Q20	83.3%	2	Q25	89.5%	2	Q30	91%	2

R: outBfsize rate between first and second partitions; P: Number of identified partitions.

Table 8

Result of partitioning BigBench queries at second phase

Query	R	P	Query	R	P	Query	R	P	Query	R	P	Query	R	P	Query	R	P
Q1	97.1%	2	Q6	85.2%	2	Q11	97.2%	3	Q16	93.7%	2	Q21	37.5%	2	Q26	90.7%	2
Q2	61.3%	2	Q7	98.2%	2	Q12	83.7%	2	Q17	98.8%	2	Q22	88.5%	2	Q27	0	1
Q3	94.4%	2	Q8	69.2%	2	Q13	61.5%	2	Q18	80.9%	3	Q23	82.4%	2	Q28	73.6%	2
Q4	45.4%	2	Q9	97%	2	Q14	97.6%	2	Q19	19.6%	2	Q24	91.9%	2	Q29	96%	2
Q5	91.7%	2	Q10	0	1	Q15	94.4%	2	Q20	83.7%	2	Q25	90%	2	Q30	90%	2

R: outBfsize rate between first and second partitions; P: Number of identified partitions.

5.4.2 Complex queries

Tables 7 and 8 show the result of partitioning BigBench queries before and after execution of the queries respectively. To execute these experiments, the configuration parameters ${Bs}^{Ve}$ and ${Bs}^{Va}$ are set to [10000 tuples/s–30000 tuples/s] and [500B–1KB] respectively, which are derived from the structure of the BigBench queries.

As shown in Table 7, for queries like Q10, the algorithm is just identified one partition with rate $\infty$ . As the $B f$ of all operators of this type of queries are Udf, the algorithm is not able to determine the average outBfsize rate of the queries before their execution so the ${Bq}_{i}^{p3}$ partition is identified for each type of these queries. The table also shows that for the most BigBench queries, the rate is high. This is due to the fact that the distance of the tuple size of the first partition operators from the second one is large. Meanwhile, for queries like Q20, the low rate may be caused by the linear structure of these type of queries. The table also demonstrates that queries like Q1 have the highest rate which is affected by the complex structure and the number of identified segments of this group of queries. Table 7 clearly shows that for the most BigBench queries, in presence of the $B s$ fluctuation, the number and type of the identified partitions at the second phase are still same with the identified partitions at the first phase. Meanwhile, for queries like Q17, the rates are different because the algorithm is able to calculate the outBfsize at the second phase.

5.4.3 Real-world queries

The result of the partitioning of the WordCount and Yahoo page load processing queries at second phase is plotted in Fig. 3. The needed configuration parameters ${Bs}^{Va}$ and ${Bs}^{Ve}$ are set to [10000 tuples/s–30000 tuples/s] and [800B–1B] respectively, which are derived from the structure of the queries.

Figure 3.

Result of two real-world queries partitioning at second phase.

As shown in Fig. 3, the rate of the Yahoo page load processing query is higher than the WordCount query because the former structure is more complex than the latter structure. Furthermore, as, there is no schema available for these queries, the algorithm has detected ${Bq}_{i}^{p1}$ and ${Bq}_{i}^{p2}$ partitions.

5.5 Bframework performance

In order to evaluate Bframework performance, we have conducted two test cases. First, we compare the average tuple latency achieved by Storm default scheduler with Bframework results under the simple and complex benchmark queries. Then, the average tuple latency achieved by a state-of-the-art scheduler [18] is compared with Bframework results under multiple real-worlds queries. All these experiments have been performed in the presence of fixed and fluctuating $B s$ scenarios. Therefore, the configuration parameter ${Bs}^{Ve}$ is set to 1000 tuples/s and [10000 tuples/s–30000 tuples/s] for fixed and fluctuating scenarios respectively. The following tables are filled with the experimental results in presence of fixed and fluctuating rate in their Fi and Fl columns respectively. Furthermore, in the following figures, Dsch, OfBsch and OnBsch stand for default scheduler, off-line Bscheduler and online Bscheduler algorithm execution curves respectively.

Figure 4.

Average tuple latency of the query Q16 of Linear Road Benchmark.

5.5.1 Simple queries

Figures 4 and 5 show the average tuple latency of the two sample Linear Road Benchmark queries during their 15 minutes execution. Furthermore, Table 9 demonstrates that the average tuple latency reduction of all Linear Road Benchmark queries. These experiments are performed by the configuration parameter ${Bs}^{Va}$ set to [4B–45B] based on the each Linear Road Benchmark query structure.

Table 9
Average tuple latency improvement under Linear Road Benchmark queries (Bframework compared with Storm default scheduler)

Q	Fx	Fl	Q	Fx	Fl	Q	Fx	Fl	Q	Fx	Fl	Q	Fx	Fl
1	3.2%	6.8%	5	5.9%	14.6%	9	4.2%	16.6%	13	5.2%	12.9%	17	3.6%	11.2%
2	2.8%	4.4%	6	6.6%	12.9%	10	6.1%	13.6%	14	6.4%	15.8%
3	7.2%	11.2%	7	7.8%	15.7%	11	7.6%	11.8%	15	6.3%	17.2%
4	1.95%	3.8%	8	8.9%	13.2%	12	15.1%	31.6%	16	14.3%	29.6%

Q: query, Fx: fixed rate, Fl: fluctuated rate.

Table 10

Average tuple latency improvement under BigBench queries (Bframework compared with Storm default scheduler)

Q	Fx	Fl	Q	Fx	Fl	Q	Fx	Fl	Q	Fx	Fl	Q	Fx	Fl
1	13%	57.2%	7	11.2%	34%	13	23.7%	65.8%	19	19.2%	52.7%	25	13.2%	49.7%
2	9.6%	38.6%	8	11.4%	27%	14	3.7%	8.15%	20	9.9%	23.5%	26	5.3%	15.6%
3	9.9%	34.6%	9	10.9%	31.1%	15	14.4%	60.4%	21	13%	54.5%	27	6.2%	9.3%
4	12%	43.2%	10	4.3%	7.6%	16	9.3%	19.3%	22	5.5%	33.6%	28	40.8%	65.7%
5	3.8%	30.6%	11	3.9%	13.4%	17	11.2%	58.6%	23	7.7%	32%	29	11.6%	47%
6	13%	54.5%	12	13%	54.5%	18	13.6%	47%	24	13%	54.5%	30	11%	36.5%

Q: query, Fx: fixed rate, Fl: fluctuated rate.

Figure 5.

Average tuple latency of the query Q8 of Linear Road Benchmark.

As shown in Figs 4 and 5, in presence of $B s$ fluctuation, the queries Q16 and Q8 show an average tuple latency reduction of 29.6% and 13.2% respectively. The improvement of Q16 is related to its structure where three segments are identified by Bframework. This segments enable the operator assignment algorithm to apply its source operator optimization and also enable the Off-line Bscheduler to assign the most determined Q16 threads to at least computing nodes. Whereas, the default scheduler does not take into account the Q16 structure and simply distributes Q16 operators. For Q8, the improvement is achieved because Off-line Bscheduler is identified two threads for ${Bq}_{8}^{p2}$ and the Online Bscheduler re-assigns them to a new computing node with lower workload at [2–3] period.

From Table 9, it is obvious to see that the Linear Road Benchmark queries have about low 14.3% average tuple latency improvement. The causes that might account this result are that the structure of these queries are generally linear, their number of operators are small and there is no inter-layer dependency. Thus, Bframework cannot apply its optimizations for these queries completely.

5.5.2 Complex queries

Figures 6–8 show the average tuple latency of the three sample BigBench queries during their 15 minutes execution. The average tuple latency reduction of all BigBench queries is presented in the Table 10. These experiments are performed by the configuration parameter ${Bs}^{Va}$ set to [500 B–10 KB] based on each BigBench query structure.

Figure 6.

Average tuple latency of the query Q18 of BigBench.

Figure 7.

Average tuple latency of the query Q1 of BigBench.

Figure 8.

Average tuple latency of the query Q28 of BigBench.

As shown in Figs 6–8, in presence of $B s$ fluctuation, the queries Q18, Q1, and Q28 have the 47%, 57.2% and 65.7% average tuple latency improvement respectively. The tuple latency improvement of Q18 is related to its identified segments and its number of operators in the first partition. The Off-line Bscheduler identifies three computing nodes to reside the first partition operators and assigns the most ${Bq}_{i}^{p1}$ operators in each segment to a distinct computing node while default scheduler assigns the operators to the computing nodes evenly. For Q1, the improvement is achieved by the Q1 second partition structure. As ${Bq}_{i}^{p2}$ structure encompasses more than 10 operators and there is an inter-layer dependency between the operators, Online Bscheduler identifies three threads and reassigned them to new computing nodes at [3–4] period which leads to 1.8% additional improvement. In addition, since there is an inter-layer dependency between ${Bq}_{i}^{p2}$ operators, the Online Bscheduler reassigns the inter-layer operator to a new related computing node at [6–7] period which leads to more 8% improvement. The query Q28 has achieved its improvement at [3–4] period because the Online Bscheduler estimates its workload and traffic load of the Q28 operators at runtime and assigns them to the appropriate computing nodes as first or second partition operators. In comparison with the default scheduler, the default scheduler does not react to the runtime traffic and work load of the Q28 and the assignment is only done based on the number of Q28 operators in presence of fixed and fluctuation scenarios.

As shown in Table 10, the achieved improvement for the BigBench queries is vary from 3.7% to 65.7% which might be caused by three main reasons. First, the structure of these queries encompasses the different styles (i.e., simple linear style or more complex star and diamond patterns or even a mix of them). For simple linear queries like Q2 the most Bframework optimizations are not applicable while Bframework optimizations can be completely applied to complex queries like Q13. Second, the number of query operators is significantly different. For some queries like Q19, there is more than 30 operators per layer which lead to the better performance improvement achieved by Bframework in the second phase. Third, the variance tuple size ${Bs}^{Va}$ of these queries can lead to their variable tuple latency improvement. For example, the tuple size of Q6 is about 2KB larger than the Q5 which increase traffic load incurred by Q6. Therefore, traffic load reduction techniques of Bframework leads to better tuple latency reduction.

5.5.3 Real-world multiple queries

In the experiment, three BigBench queries including linear, star, diamond styles and two real-world queries including WordCount and Yahoo page load processing accepted simultaneously. The experiment is performed by the configuration parameter ${Bs}^{Va}$ set to [300KB–400KB] based on each query structure. Figure 9 shows the average tuple latency of the queries during their 15 minutes execution. Note that, in Fig. 9, State stands for the state-of-the-art scheduler.

Figure 9.

Average tuple latency improvement under real world multiple queries (Bframework compared with state-of-the-art scheduler).

As shown in Fig. 9, Bframework outperforms the state-of-the-art scheduler in tuple latency reduction up to 31%. The improvement is related to the difference between Bframework strategies with state-of-the-art scheduler for scheduling multiple accepted queries. In other words, for two accepted queries ${Bq}_{i}$ and ${Bq}_{j}$ , state-of-the-art scheduler sorts all ${Bq}_{i}$ and ${Bq}_{j}$ operators in descending order by their traffic load and assigns them orderly to computing nodes with lighter work load. This strategy can lead to a situation where a heavy traffic load operator of ${Bq}_{j}$ is scheduled before a ${Bq}_{i}$ light traffic load operator which may lead to an increase of waiting time of ${Bq}_{i}$ operators. Whereas, Bframework schedules the operators of ${Bq}_{i}$ independently of the ${Bq}_{j}$ operators and the ${Bq}_{j}$ operators do not increase the waiting time of ${Bq}_{i}$ operators.

5.6 Discussion

According to the conducted experiments and their analysis, we can conclude that Bframework is able to correctly partition the streaming Big Data analytic queries based on its input Big Data stream. Experimental results show that Bframework is able to correctly partition simple and complex before the execution of queries and can also be used in real world streaming Big Data analysis environment.

Furthermore, Bframework outperforms the Storm default scheduler in the average tuple latency reduction under the simple and complex streaming Big Data analytic queries up to 29.6% and 65.7% respectively. Experimental result have shown that Bframework is able to reduce the average tuple latency of simple queries mainly through the appropriate assignment of computing nodes. For complex queries, Bframework can also be used to efficiently reduce the average tuple latency through the efficient scheduling and rescheduling of needed threads, processes and computing nodes.

In addition, compared to the state-of-the-art scheduler, Bframework is able to reduce the average tuple latency of multiple accepted real world queries up to 31%. Bframework is able to efficiently schedule simultaneous accepted queries through isolated assignment of the query operators to the specific computing nodes.

Finally, Bframework performs best in complex fluctuating streaming Big Data analytic queries. Experimental results prove that most Bframework optimization techniques are completely utilized by queries with complex structures and fluctuating input Big Data streams.

6. Conclusion

The demand for streaming Big Data analysis applications is increasing. As the demand is growing, the efficient execution of the applications using suitable platforms become more important. However, current platform schedulers are not suitable for these applications in an effective manner. This study presents Bframework, a framework designed to efficiently schedule the accepted streaming Big Data analytic queries on a resource cluster. The five modules of Bframework implement the proposed operators profiling and partitioning, operator assignment, Off-line Bscheduler and Online Bscheduler algorithms and operate prior to and during the execution of accepted queries. Bframework is developed based on the idea of using more than one scheduling strategy for streaming Big Data analytic queries. Bframework partitions a query based on its Big Data stream characteristics and applies a set of efficient scheduling and rescheduling strategies for each partition prior to and during the execution of query.

Experimental results show that, compared to Storm’s default scheduling, Bframework is able to efficiently reduce tuple latency of streaming Big Data analytic queries up to about 65%. Furthermore, Bframework outperforms the most recent scheduling study under real-world queries in tuple latency reduction by up to 31%.

In the future work, it is planned to extend Bframework on public cloud resources. An efficient resource provisioning mechanism will be considered in our following work. Moreover, experiments that executes Health Big data queries will also be included.

References

Arasu

Cherniack

Galve

Maier

Maske

A.S.

Ryvkina

Stonebraker

and Tibbett

, Linear road: a stream data management benchmark, in: Proceedings of the Thirtieth International Conference on Very Large Data Bases, VLDB Endowmen, Vol. 30, 2004, pp. 480–491.

Ghaza

Rab

Raa

Poess

Crolotte

and Jacobse

H.A.

, BigBench: towards an industry standard benchmark for big data analytics, in: Proceedings of the 2013 ACM SIGMOD International Conference on Management of Dat, 2013, pp. 1197–1208.

Safae

, Real-time processing of streaming big dat, Real-Time Systems 53(1) (2017), 1–4.

Apache Storm, Available from: https://github.com/apache/storm [Accessed 5

{}^{\text{th}}

June 2016].

Zhang

and Luo

, Real-time scheduling based on optimized topology and communication traffic in distributed real-time computation platform of storm, Journal of Network and Computer Application, 2017.

Abad

D.J.

Carne

Çetinteme

Cherniack

Conve

Lee

Stonebraker

Tatbu

and Zdonik

, Aurora: A new model and architecture for data stream management, The VLDB Journa – The International Journal on Very Large Data Bases 12(2) (2003), 120–139.

and Huang

, A stable online scheduling strategy for real-time stream computing over fluctuating big data stream, IEEE Access, 4 (2016), 8593–8607.

Zhang

and Zheng

, Building a fault tolerant framework with deadline guarantee in big data stream computing environments, Journal of Computer and System Sciences, 2017.

Wen

Chu

T.S.

and Xuelong

, Toward scalable systems for big data analytics: A technology tutorial, IEEE Access 2 (2014), 652–687.

10.

Che

Tan

and Su

, T-storm: Traffic-aware online scheduling in storm, in: Distributed Computing Systems (ICDCS, 2014 IEEE 34𝑡ℎ International Conference on, 2014, pp. 535–544.

11.

Berman

J.J.

, Principles of big data: preparing, sharing, and analyzing complex information, Newnes, 2013.

12.

Fegaras

, Incremental query processing on Big Data stream, IEEE Transactions on Knowledge and Data Engineering 28(11) (2016), 2998–3012.

13.

Barlow

, Real-time big data analytics: Emerging architecture. O’Reilly Media, Inc., 2013.

14.

Chen

Mao

Zhang

and Leung

V.C.

, Big data: related technologies, challenges and future prospects, Heidelberg: Springer, 2014.

15.

Kaur

and Soo

S.K.

, Dynamic resource allocation for big data streams based on data characteristics (5Vs), International Journal of Network Management, 2017.

16.

Kaur

and Soo

S.K.

, Efficient Resource Management System Based on 4Vs of Big Data Streams. Big Data Research, 2017.

17.

Huan

and Sun

, Analysing and evaluating topology structure of online application in Big Data stream computing environmen, International Journal of Wireless and Mobile Computin 10(4) (2016), 317–324.

18.

Dwarakanatha

, S-Flink: Schedule for QoS in Flink Using SD, in: Computer Software and Applications Conference (COMPSAC), 2016 IEEE 40th Annual, Vol. 2, 2016, June, pp. 620–621.

19.

Kulkarni

Bhaga

Kedigehalli

Kellogg

Mitta

Patel

J.M.

Ramasamy

and Tanej

, Twitter heron: Stream processing at scale, in: Proceedings of the 2015 ACM SIGMOD International Conference on Management of Dat, 2015, May, pp. 239–250.

20.

De Mattei

and Mencagl

, Keep calm and react with foresight: Strategies for low-latency and energy-efficient elastic data stream processin, in: Proceedings of the 21st ACM SIGPLAN Symposium on Principles and Practice of Parallel Programming, 2016, p. 13.

21.

Tan

and Xu

, A predictive scheduling framework for fast and distributed stream data processing, in: Big Data (Big Data), 2015 IEEE International Conference on, 2015, October, pp. 333–338.

22.

Trident API overview, Available from: storm.apache.org/releases/1.0.0/Trident-API-Overview.html [Accessed 7

{}^{\text{th}}

June 2016].

23.

Cardellin

Grass

Lo Presti

and Nardell

, Distributed QoS-aware scheduling in Storm, in: Proceedings of the 9𝑡ℎ ACM International Conference on Distributed Event-Based Systems, 2015, pp. 344–347.

24.

Cardellin

Grass

Lo Presti

and Nardelli

, Optimal operator replication and placement for distributed stream processing system, ACM SIGMETRICS Performance Evaluation Revie 44(4) (2017), 11–22.

25.

Cardellin

Grass

Lo Presti

and Nardell

, Optimal operator placement for distributed stream processing applications, in: Proceedings of the 10𝑡ℎ ACM International Conference on Distributed and Event-based System, 2016, pp. 69–80.

Efficient resource scheduling for the analysis of Big Data streams

Abstract

Keywords

1. Introduction

1 storm.apache.org.

2.1 DSP scheduling

2.2 Streaming Big Data scheduling

3. Problem statement

3.1 Big Data stream

3.2 Streaming Big Data analysis query

4.1 Bframework architecture

4.2.1 Query profiling and partitioning algorithm

Table 1 Simple selectivity criteria

4.2.4 Online Bscheduler

4.3 Bframework algorithms complexity analysis

5. Evaluation

Table 2 Characteristics of BigBench queries [1]

5.2 Environment

Table 3 Storm cluster specifications

Table 4 Parameters used in Bframework

Table 5 Result of partitioning Linear Road Benchmark queries at first phase

Table 7 Result of partitioning BigBench queries at first phase

5.4.3 Real-world queries

Table 9 Average tuple latency improvement under Linear Road Benchmark queries (Bframework compared with Storm default scheduler)

6. Conclusion

References

¹
storm.apache.org.

Table 1
Simple selectivity criteria

Table 2
Characteristics of BigBench queries [1]

Table 3
Storm cluster specifications

Table 4
Parameters used in Bframework

Table 5
Result of partitioning Linear Road Benchmark queries at first phase

Table 7
Result of partitioning BigBench queries at first phase

Table 9
Average tuple latency improvement under Linear Road Benchmark queries (Bframework compared with Storm default scheduler)