Big data analytics have become a necessity to businesses worldwide. The complexity of the tasks they execute is ever increasing due to the surge in data and task heterogeneity. Current analytics platforms, while successful in harnessing multiple aspects of this \data deluge", bind their e cacy to a single data and compute model and often depend on proprietary systems. However, no single execution engine is suitable for all types of computation and no single data store is suitable for all types of data. To this end, we present and demonstrate a platform that designs, optimizes, plans and executes complex analytics work ows over multiple engines. Our system enables users to create work ows of variable detail concerning the execution semantics, depending on their level of expertise and interest. The work ows are then analysed in order to determine missing execution semantics. Through the modelling of the cost and performance of the required tasks over the available platforms, the system is able to match distinct work ow parts to the execution and/or storage engine among the available ones in order to optimize with respect to a user-de ned policy.
Big data analytics have become indispensable for the
majority of industries, enabling engineers, analytics experts
and scientists alike to tap the potential of vast amounts of
business-critical data. Such data analysis demands a high
degree of parallelism in both storage and computation and
has given rise to diverse execution engines and data stores
that target speci c data and computation types.
Performance optimizations thereof assume strictly single-engine
environments, thus considering speci c data formats and
query/analytics task types [
However, modern work ows have become increasingly long
and complex and may include multiple data types such as
relational, key-value, graph, etc., as well as greatly diverse
operators, ranging from simple Select-Project-Join (SPJ)
and data movement to complex NLP-, graph- or custom
business-related operations. What is more, they are
executed under varying constraints and policies (e.g., optimize
performance or cost, etc.). There currently exists no single
platform that can optimize for this complexity [
Sensing this trend, companies now o er cloud software
distributions that incorporate di erent processing frameworks,
data stores and libraries to alleviate the burden of
multiple installations and con gurations [
To address multi-engine analytics work ow optimization we demonstrate an integrated, open source platform for creating, managing, executing and monitoring complex analytics work ows1. Its goal is twofold: (a) To allow the user to design work ows that span multiple engines and data stores by either giving speci c details on execution semantics of tasks and data stores or leaving the platform to determine them through an automated work ow analysis phase. Then, the work ow goes through an automated optimization phase, before being sent for execution. (b) To provide costbased and customizable resource management of the diverse execution and storage engines available by adaptively choosing to execute each sub-part of the work ow to a (possibly di erent) deployed engine in order to enhance performance
Towards this direction, our system supports the most prevalent open-source execution models (e.g., Map-Reduce, Bulk Synchronous Parallel) as well as state-of-the-art centralized and distributed storage engines (RDBMSs, NoSQL, distributed le-systems, etc.) and is able to optimize work ows consisting of tasks that range from simple group-by, aggregation or complex joins between di erent data sources to machinelearning tasks.
Our demonstration will showcase our system's ability to i)
model operator performance according to di erent engines
and their resources and ii) adaptively decide on which
operator version to run based on the optimization policy and the
available engines. The demonstration platform will integrate
Hadoop [
user policy
Interface r e z y l a n A parse match refine validate
Making r e z iim Planner t p O r o t u c e x E
policy and input aspects, while being able to evaluate the advantages of multi-engine optimization by inspecting generated plans and output.
The system integrates two core components of the ASAP
project, namely PAW (Platform for Analytics Work ows),
which acts as the work ow management tool of ASAP and
IReS (Intelligent Resource Scheduler) [
The components of PAW communicate using the internal work ow representation and are:
Interface. The interface enables users to interactively create and/or modify a work ow.
Analyzer. The analyzer parses the work ow, identi es operators and data stores and maps them to existing operator implementations, generates metadata of edges, nds edges where the data conversion should be applied and adds the appropriate conversions. The operator implementations reside in the IReS operator library. The operators are classi ed as, either analytics operators, which perform the core analytics jobs over the data, or the associative operators, which serve as `glue' between di erent engines and perform move and transformation operations. Optimizer. The optimizer generates a functionally equivalent work ow, optimized for performance objective.
After the generation of the optimized work ow, PAW hands it to the IReS platform for further cost-based optimization according to the use-de ned policy, planning and execution. IReS receives from PAW the input that is necessary for its operations and identi es execution artifacts such as operators, data, their dependencies and accompanying metadata. Moreover, it validates the user-de ned policy. All this information must be robustly identi ed, structured in a dependency graph and stored.
IReS is decomposed in two layers, the cost-based optimizer and the executor layer: The cost-based optimizer layer is responsible for optimizing the execution of an analytics work ow with respect to the policy provided by the user. The core component of the optimizer is the Decision Making module, which determines the optimal execution plan in real-time. This entails deciding on where each subtask is to be run, under what amount of resources provisioned, the plan for moving data to/from their current locations and between runtimes (if more than one is chosen) and de ning the output destinations. Such a decision must rely on the characteristics of the analytics task in hand and the models of all possible engines. These models are produced by the Modeling module and stored in a database called Model DB. The initial model of an engine results from pro ling and benchmarking operations in an o ine manner, through the Pro ling module. This module directly interacts with the pool of physical resources and the monitoring layer in-between. While the work ow is being executed, the initial models are re ned in an online manner by the Model re nement module, using monitoring information of the actual run. Such monitoring information is kept in the IReS DB and is utilized by the decision making module as well, to enable real-time, dynamic adjustments of the execution plan based on the most up-to-date knowledge. The executor layer is the layer that enforces the optimal plan over the physical infrastructure. It includes methods and tools that translate high level \start runtime under x amount of resources", \move data from site Y to Z" type of commands to a work ow of primitives as understood by the speci c runtimes and storage engines. Moreover, it is responsible for ensuring fault tolerance and robustness through real-time monitoring.
Both PAW and IReS are open source2. The interface of
PAW is a web application in Jade [
The demonstration shows how PAW and IReS
collaborate in order to optimize an analytics work ow, produce an
execution plan that conforms to a user-de ned policy and
enforce it. We show that the system can be used by users
with di erent level of expertise in order to create work ows.
The tool implements a novel work ow language [
The system is controlled by a comprehensive web-based GUI that attendees will utilize. The GUI controls a cloudbased deployment of several runtime engines and data stores over 16 virtual machines of an Openstack cluster.
The users will be able to test the system either using one of our prede ned work ows or assembling their own, using operators from the ASAP operator library. A diverse set of operations of varying complexity is covered including basic SQL queries (selections, projections, joins) and ML algorithms (classi cation and clustering).
The prede ned work ows represent real use cases driven by business needs. These cover complex data manipulations in the areas of business analytics on telecommunication data and web data analytics, provided by a large telecommunications company and a well-known web archiving organization respectively. The input datasets for these work ows consist of anonymized telecommunication traces and web content data (WARC les). Subsets of those datasets can be used for each of the available work ows.
This example shows the telecommunication use case of ASAP. This work ow detects peaks representing an event, by comparing the density of population within a region in a given moment against the expected density for that area at that hour of the day. It involves processing of anonymized Call Detail Records (CDR) data (residing in an RDBMS) via clustering along time and space in order to detect peaks in load according to a set of criteria. The dataset of peaks is used to discover clusters of calls that occur with or without regularity. Figure 2 displays the work ow.
The rst steps of the process, data prep and convert ts, consist of call aggregation by user id and extraction of time periods (for instance, days) and smaller time slots (for instance, hours). The regions table contains links of tower ids and regions. Time slots and regions of analysis, which are used in lter region, are parameters provided by the user. These two parameters, then, allow de ning a spatiotemporal grid, and each observation of an input dataset can be assigned to one of its cells. The number of observations that fall in a cell de nes its density. The input data is partitioned into two sets: a training dataset (for instance, 1st month) and a test dataset (for instance, left time period), two lter s following after convert ts produce them. For both datasets the spatiotemporal grid of densities is computed. The rst is used to compute the densities of a typical period for each region. The second dataset is then compared against such typical period in order to detect signi cant deviations. Based on the densities obtained for each region and each time slot over the training dataset, an expected density value is computed for each region, by averaging the densities (week aggr ation) measured at the same time slot of all the periods in the time window covered by the dataset. For instance, we might obtain an expected density for each pair (region, hour of the day), i.e., 24 values for each region, assuming 24 one-hour time-slots. Then, for each region and each time-slot, the corresponding density is compared against its expected value (join and calc ration): if the difference is signi cant ( lter peaks), an event of form (region, weight, time slot) is produced (events), representing its spatiotemporal slot and a discretized measure (weight) of how strong was the deviation.
Figure 3 displays the optimized work ow produced by PAW's Optimizer. In this there are tasks pushed closer to the start of the work ow and merged in a single node, these tasks are in the original nodes join, lter region and regions dataset. This optimization has been done accordingly to a heuristic that moves restrictive operators (in this case lter region) to the root of the work ow. Furthermore the Optimizer of PAW checks that this rearrangement and merging can be performed.
This work ow is then handed to the IReS platform, where the conceptual work ow operators are matched with implementations that exist in the operator library, creating all alternative execution paths. After consulting the cost models of the implemented operators, the platform chooses the plan and decides on the exact engines and their setup in order to optimize the work ow in terms of the user-de ned policy. In our running example, if the input dataset is large the decision making module opts for a plan that contains distributed implementations of the operators in Spark rather than centralized ones in PostgreSQL. This choice entails the transfer of the dataset from PostgreSQL (where it originally resides) to HDFS, thus an auxiliary move operator is added to the work ow. Moreover, assuming the user only cares to minimize execution time, IReS decides to utilize all available nodes of the cluster.
This work has been supported by the European Commission in terms of the ASAP FP7 ICT Project under grant agreement no 619706. Nikolaos Papailiou has received funding from IKY fellowships of excellence for postgraduate studies in Greece - SIEMENS program.