High-level declarative scripting languages are considered to be an effective tool for specification and execution of complex analytical data processing scenarios as they can hide the complexity of underlying heterogeneous processing environment. A growing demand to process huge amounts of data under real-time requirements as well as advanced similaritybased models raises the need in approximate processing. We discuss architecture of an extendable system for optimization and execution of approximate complex queries formulated in similarity-based declarative query language.
Complex querying scenarios appear in various data integration and processing tasks. Several different information resources, such as databases, semi-structured data, streams, and uncertain data might be needed in the same data processing workflow. The systems might be heterogeneous in terms of data model, dynamics, trustfulness, and content type, as well as querying or retrieval paradigms. The need to combine data extracted from heterogeneous resources potentially based on diverse querying paradigms appears in several application contexts and application areas, including advanced search, personalization, and analytical processing.
Specifications of complex analytical scenarios can be effectively expressed in high-level declarative scripting languages hiding the complexity of underlying processing environment.
The systems integrating these ideas usually exploit intermediate algebraic languages which in addition to common features of query languages, such as filtering, settheoretic operations, and joins incorporate their fuzzy extensions and complex processing techniques such as NLP, mining, and analytics. Any complex data workflow processing based on similarity or other uncertain querying model tends to be com
This work was partially supported by HP Labs under contract No. CW317234. putationally heavy. The expectation is that a declarative algebraic approach to query evaluation over diverse set of information resources enabling query optimization is capable to address this performance issue.
Query optimization techniques for such context differ from relational due to significant differences in algebraic properties and cost models for extended similarity-based operations.
The uncertain nature of the queries implies a need in approximate query evaluation because in such context exact query evaluation sometimes is not reasonable or meaningful.
The approximate algorithms implementing operations of the algebra in combination with traditional exact query evaluation can be applied to reduce the query evaluation time. Some algorithms may produce different quality of the output depending on the amount of resources allocated for execution. This leads to the analysis of speed/quality trade-offs in the context of query optimization and evaluation.
The problem of traditional query optimization, namely “find an execution plan with the minimal cost” is replaced with either “find the best plan yielding at least specified quality”, or “provide the best possible quality for at most given amount of resources”.
We have to maximize the quality having the limited resources for query evaluation. Thus, compared to traditional query optimization the system incorporates the resource allocation step in the whole query evaluation pipeline.
Our primary objective is to build an architecture which is able to process complex data processing workflows in real-time, that is ensure predictable response time controlling approximate evaluation. 1.2
The example from model implementation is based on data from OpinRank dataset with judgments. The dataset contains hotels from 10 cities over the world. For each hotel its name, address, and scores on cleanliness, room, service, etc. are available.
Let us consider a query: find best 10 hotels in London
according to service and cleanliness. The query is
formulated in terms of high-level declarative query language
similar to one proposed in [
In this work we further do not distinguish operation and algorithm its implementing because most modules in the architecture process algorithms rather than logic operations.
The primary operations retrieve hotels with scores from primary sources based on provided score types. Subsequent top operations are applied to reduce the number of objects for further analysis. The next price filter retrieves the price for double room for specific date. The city filter keeps only hotels in London. The fusion operation combines scores for hotels based on provided rule. Finally, top operation is applied to retrieve only 10 best hotels according to new input scores.
(a) Example query evaluation plan (b) Alternative query evaluation plan (c) Approximate query evaluation plan
One may see that there are several opportunities for optimization. For example, we can swap unary operations. The system also supports implementation of fusion operation which simultaneously retrieves best objects and integrated implementation of primary operation and city filter.
Figure 1(b) demonstrates how the query evaluation plan can be transformed (not necessarily making it optimal).
The quality of the query evaluation result depends on the amount of time specified by the user.
Thus, as soon as the query optimization step produces
a reasonably good query evaluation plan the system has
to distribute fixed amount of resources specified by the
user. Figure 1(c) shows some query evaluation plan
with operations admitting approximate implementation
marked by dashed borders. Approximate algorithms for
them and corresponding cost models are described in
details in [
Let us suppose that we analyze social sources (such as ratings from retailer web site and customer tweets) to find out how customer satisfaction depends on price, for each product group. Thus, we probably formulate the following query: count average values of ratings from retailer and from sentiments, grouped by price range, for each product group. The corresponding query evaluation plan is shown in figure 2 with the following description of nodes: 1: group by PriceRange
nest (ProdGroup, avg(Rating),avg(Sent)) 2; group join on ProductModel 3: group join on ProductModel 4: get database product table 5: get retailer ratings for products 6: get sentiments from tweets for products (a) Possible resource allocation (b) Another resource allocation
Let us as well suppose that external primary sources produce results as output streams of objects with different speed: 500 objects per second for retailer ratings and 300 objects per second for sentiments extracted from tweets. As the data sources are actually (potentially unlimited) streams, an exact evaluation is impossible. We assume that an approximate result is expected in at most 230 ms. This implies an ultimate need in approximate evaluation which restricts the quantity of data items processed by each of operations.
The limited amount of resources can be distributed among operations in the plan in different ways shown in figures 2(a), 2(b). One may see that the resource allocation strategy shown in figure 2(b) is preferable because we do not retrieve excessive tweets when have not enough retailer ratings. We propose an architecture for an extendable system for optimization and execution of approximate complex queries formulated in similarity-based declarative query language. Based on the proposed architecture we demonstrate the integration of techniques proposed in previous research and connect together several parts: similaritybased algebra, extended cost/quality models, and resource allocation techniques. We provide the detailed discussion of several aspects of the architecture focusing on its extensibility.
The rest of the paper is structured as follows. Section 2 describes the architecture including underlying data model, intermediate algebraic query language, and main modules. The architecture details on specific modules are followed by a detailed discussion of classes of operations from the system extensibility point of view in section 3. Section 4 outlines the related work. 2
This section outlines high-level architecture of a middleware query processing system acting on top of existing query evaluation and data processing systems, and below some user interface for query formulation and results exploration.
The system supports different types of data sources and applies the uniform representation of different types of data to enable their integration in one query. This results in a compromise between uniform integration of different types of sources and processing units and necessity to control the query correctness and validity at the query formulation step. 2.1
The central concept of our model is a q-set defined in
[
The algebra on q-sets is characterized in [
Query processing engine contains modules shown in figure 3: parser, optimizer which interacts with transformations and cost models, resource allocation module which is based on quality models, and executor. The parser translates a query into the initial query evaluation plan that is the query tree with algebraic operations.
The initial query is supposed to be formulated in some
high-level language: extension of SQL similar to one
proposed in [
The optimizer in query evaluation system in the described environment should be adaptive and distensible.
Similar to the traditional optimization of relational queries the optimizer has to analyze the algebraic properties of underlying operations. The associativity and distributive properties enable the rewriter to form the space of query plans based on the transformations of the query tree.
We have to support algebra extensibility and develop optimizer tunable to new specific operations which may possess new algebraic equivalences and expand the space of possible query plans.
The optimizer is based on extensible and tunable set of possible transformations of the query representing the space of query evaluation plans. It is important to note that in the proposed architecture the set of transformations includes all possible reorganizations of the query evaluation (sub) plan: based on algebraic equivalencies or different implementations of algebraic operations. This fact enables an easy way to introduce new operations, algorithms, and corresponding transformations.
In comparison with traditional relational data bases the space of transformations is not as homogeneous as in extended algebra: for example, non-associative joins and complex transformations with top k operation are there.
A special attention on the implementation of transformations in the system is needed because of its extensibility: the space of transformations should not be deeply coded in the optimizer, thus uniform interfaces are needed to make transformations easily pluggable in.
Any high-quality optimizer should be cost-based, hence cost models for all operations should be provided, including user-defined extensions. In order to be able to select a good query evaluation plan the optimizer for complex query processing should be constructed based on cost models of operations in the queries. The architecture should support an integration of new cost models for operations and tuning of existing ones. The cost models are supposed to be tunable and adaptive, that is, be able to use data statistics when available.
In the model implementation an optimizer is based on limited descent by plan cost.
Further the adaptive query optimization technique
[
The architecture supports the class of approximate algorithms with the following properties: with controllable quality (allocated resources define the quality of result); with fixable parameters (the fixed amount of allocated resources is transferred into fixed operation call parameters).
The restricted class of supported approximate
algorithms contains quite large number of specific
algorithms: for example, an approximate algorithm for
aggregation operation proposed in [
Approximate algorithms with fixable parameters
should be distinguished from any-time techniques when
user is able to stop operation evaluation based on
estimations of already achieved result quality, for example
proposed in [
Approximate pre-configurable algorithms are more restrictive as the estimation of result quality should be done in advance. On the other hand, this property is less restrictive as there is no need to support correct approximate result of operation evaluation at any time of execution.
As the focus of this work is on the approximate query
evaluation the cost models are extended with the
notion of quality and form the quality models for algebraic
operations. Few examples of such extended cost
models are presented in [
Although the semantics of data quality is
complex [
It is important to note that cost/quality models are based on relative operation quality. Let us consider unary operation which receives input data with absolute quality ain. Even the best possible implementation of operation may reduce the quality of the result data aout. Thus, the quality of operation equals aaoiunt . If the operation receives the unlimited resources for the fixed input data it produces data with the best feasible absolute quality aout(1): If it receives amount of resources t; which is less than needed for the best quality, the absolute quality of its output is aout(t); thus, the relative quality of operation is aaoouutt((1t)) :
Another concept essential for our extended cost/quality models is a resource. Different types of resources may be used to restrict the query execution, some of them, like CPU time, CPU cycles, and I/O, depend mostly on the complexity of the plan, while others like memory size or elapsed time may also depend on available configuration (e.g. number of processors allocated for the query).To treat all kinds of resources uniformly, we consider configuration to be a part of the plan.
In this work the amount of resources is expressed with single numeric value. This value may represent either the most important type of resources, e.g. elapsed time for real-time requirements, or a combination of different resource types.
In order to construct the query evaluation plan which operates in the limited amount of time and provides the best possible quality the resource allocation module should interact with the optimizer. The optimization problem for approximate query evaluation seems to be much harder than commonly known exact query optimization: in addition to selection of one of equivalent execution plans, the optimizer has to distribute available processing resources between operations of the query execution plan.
Model implementation uses an approximate solution
described and analyzed in [
Obviously, the above solution may result in suboptimal plan. For example, approximate evaluation of operations may limit the cardinality of intermediate results making the selected query evaluation plan inefficient. Another limitation is in the inability to support approximate algorithms implementing operations without controllable quality but still relevant for approximate real-time query evaluation. An integrated implementation of an optimizer and resource allocator is planned for future work. 2.2.4
The executor is based on commonly accepted pipeline query evaluation augmented with provisions for extendibility. Thus intermediate results of evaluation are transferred between operations without materialization; however, some implementations of specific operation still may materialize intermediate results internally.
The set of operations can be extended by other generic or user operations; new exact and approximate algorithms implementing operations also can be smoothly introduced to the system together with new algebraic equivalences and cost/quality models. 3
The main requirements to the system are extensibility and approximation, thus we should take them into account in the system architecture. 3.1
A notable feature of the environment in consideration is the need to provide for configurable transformations, in contrast with usually hard-coded transformations for relational optimizers.
Transformations are described in the system by 2 functions and are registered in the list of available transformations.
Check function which evaluates the applicability of transformation to the root of the query evaluation (sub) plan.
Apply function which constructs a new query evaluation (sub) plan from input one applying the transformation.
The system has base implementation of some universal query transformations such as algorithm change transformation and distributive property. This enables easy implementation of additional transformations just configuring base ones. 3.2
As any effective query optimization is based on cost models of operations the system should support userdefined cost models for operations to be able to improve applied cost models for basic operations and add new ones for extensions.
A cost model is a function with the following interface: Input: Operation call parameters; input data statistics. Output: Operation cost; output data statistics. 3.3
At the resource allocation step the system applies quality models which describe the trade-off between cost of operation execution and quality of result for approximate algorithms.
A quality model is a function with the following parameters: Input: Input data statistics; non-changeable operation call parameters.
Output: Piecewise-linear representation of dependence between resources allocated for operation execution and result quality.
In addition, a function mapping allocated resources and quality into operation parameters is needed for correct work of the system. This function returns operation call parameters based on specified operation execution conditions (resources, quality or both): Input: Input data statistics; non-changeable operation call parameters; resources allocated for operation execution; expected operation result quality.
3.4
Discussion of operation library is followed by architecture details for specific classes of operations: primary, unary, and binary focusing on algebra extensibility. 3.4.1
Operation library stores the set of algebraic operations and all attendant structures. For each operation the library stores the following:
Mapping of operation call in the query evaluation plan into the direct call;
Relevant transformations using the operation.
Each algorithm, for example join based on nested loops or sort merge, is registered in the operation library as a separate operation. To add new operation to the operation library one should register the corresponding cost model, quality model, and transformations.
From the architecture point of view the set of algebraic operations can be divided into: primary operations, unary operations, binary operations.
During the algebra extension operations from all these groups can be implemented supporting predefined interface discussed below or constructed based on universal basic implementations for each class.
Parameters of data processors and functions used to configure basic implementations are passed from operation parameters: map of parameter names into values. In this case the interface of basic implementation is universal to some extent and different data processors can work with different sets of specific parameters.
All operations support the following interface: Input: Operation call parameters, pipelined argument(s)
The number of arguments depends, of course, on the arity of the operation. 3.4.2
Primary sources of data are implemented as operations without input arguments.
For example, if the primary source is a database, it is implemented as an operation with input parameters representing SQL query to retrieve data and data needed to setup the connection.
A built-in basic implementation of primary operation is provided. It can be configured by data retrieval processor to construct different real primary operations.
Data retrieval processor returns an object from data set for each call until all data is returned. Basic implementation of primary operation calls data retrieval processor while the latter returns objects and transfer collected objects to the output pipe. To extend the algebra by new primary operation one may implement a data retrieval processor and register it in the system. After that a new primary operation may be implemented as basic primary operation configured by new data retrieval processor can be registered in the operation library. 3.4.3
The architecture supports the basic implementation of the unary operation which can be configured by unary data processor. Basic implementation of unary operation reads objects from input pipe one by one, asynchronously calls unary data processor, and when results of processing are available it forwards objects to the output pipe.
Unary data processor implements 2 methods: put object, get object. Put method retrieves object and processes it in specific way, for example adds new attribute or inserts object into the internal list of object sorted by score. Get method returns the results of processing (objects) one by one if they are available in the moment of method call.
Let us consider the architecture of unary operation based on aggregation. Aggregation process can be implemented as 4 unary operations: grouping operation which defines how objects are grouped, that is adds group label to each object; aggregating operation, which constructs objects representing groups based on input objects with group labels; group score calculation operation, which retrieves input objects representing groups and returns them with scores calculated based on specific rule; aggregate calculation operation, which retrieves input objects representing groups and returns them with additional attribute (aggregate) calculated based on group of objects, for example sum on some attribute.
All these operations can be implemented using the basic one configured by corresponding unary data processors. 3.4.4
Binary operations retrieve objects from 2 input pipes and return result objects when possible to the output pipe.
The system architecture supports the basic implementation of the binary operation based on construction of cross product of operation arguments. Basic implementation is configured by predicate function which takes two objects as arguments and returns the object constructed based on input pair of objects with corresponding score. Basic implementation of binary operation reads objects from input pipes, calls predicate function, and puts to the output pipe new objects with scores from predicate function.
Thus the extensibility of algebra by new binary operations can be based on implementation of new predicate functions. It is important to note that basic configurable predicate implementations can be applied: predicate function comparing attribute value with constant takes attribute name, constant value, and sing of comparison as parameters.
New binary operations can be implemented without predicate function supporting interface used by basic implementation. It is especially important in case of implementation of approximate algorithms for binary operations. One can extend the core of the system with other configurable basic implementations of binary operation, for example approximate algorithms based on partial consideration of cross product pairs. 4
A comprehensive discussion of the related work can be
found in [
Several data processing systems are developed to
process Big Data [
The system for optimization and procession of
complex analytical workflows is discussed in [
System for approximate evaluation of SQL
aggregation queries is proposed in [
Systems presented in [
Traditional data querying means are reconsidered and
extended because of variety of data under processing.
Proposed declarative querying languages are mostly the
extensions of relational algebra [
Some extensions of querying languages are based on
extension of relational data model: attribute
representing rank or score of object is marked out from the set of
attributes [
Based on the extended data model new algebraic
operations can be introduced. Several types of querying
language extension can be outlined:
languages supporting user-defined weights
representing the importance of data source or sub-query
[
The ideas and approaches for querying systems with
different data processing paradigms, such as relational
database and information retrieval in collection of
documents, were considered in [
A brief overview of classical query optimization
techniques can be found in [
Authors of [
Extended querying languages support some algebraic
equivalencies considered in [
The development of cost models is base for
optimization of queries specified in extended high-level language
[
Authors of [
The proposed system architecture is based on the
authors previous research and connects together several
parts: similarity-based algebra [
The proposed architecture of an extendable system for optimization and execution of approximate complex queries formulated in similarity-based declarative query language demonstrates the possible integration of techniques proposed in previous research and connects together several parts: similarity-based algebra, extended cost/quality models, resource allocation approach.
The future work includes full prototype development supporting adaptable cost and quality models, adaptive query evaluation, enhanced optimizer for approximate query evaluation, and support for multiple execution platforms.