We challenge the common assumption that queries are submitted to a pre-configured, already running engine and put forward the idea of dynamically instantiating a chosen data processing engine upon query submission by leveraging Function-as-aService (FaaS) platforms. We demonstrate the idea by running unmodified data processing engines (we use Apache Drill as an initial example) on real-world serverless FaaS platforms and show that such engines can be instantiated on demand when a query arrives. We aim to eventually support a wide range of queries and workloads. Wide access to such functionality would be a game changer in data processing. First, it would enable pay-per-query models supporting sporadic, interactive data analysis on arbitrary engines. Second, it would significantly increase the flexibility for data processing by enabling the possibility of dynamically choosing the actual engine, its configuration, and the resource allocation on a per-query basis. Logically, this amounts to dynamically attaching a query engine to the query rather than sending the query to a pre-configured and already deployed engine. In this paper we elaborate on this vision, outline the design of the MetaQ prototype that we are building to explore the idea, demonstrate that it is realistic through initial experiments, and discuss its many exciting practical implications.
is instantiated (potentially selected from a variety of
engines) in the best possible configuration and deployment
Operating a long-running query engine has several lim- for the query, the query is executed by the engine, and
itations. First, it generates costs even if it is idle. Sec- upon completion, the engine is shut down (unless there is
ond, most distributed query engines lack elasticity, which a reason to keep it running, like a similar query arriving
leads to deployments being over-provisioned to cope while the engine is active). This eliminates the need for
with potential peak loads [
In this paper we explore an ambitious and radically long-running engines (the engine settings need to be
opnew design: one in which we take advantage of server- timized only for the given query, which allows for more
less computing to provide ephemeral per-query engines specialized and eficient solutions [
The vision of EPQE is enabled by the emergence of SM - session manager MO - meta-system optimizer PP - platform provider
SM MO
Function as a Service (FaaS). In serverless computing, spective. These systems propose, among other things,
users deploy and invoke fine-grain functions on-demand [ 9, complex ways to reduce the overhead of communicating
10]. There are three main characteristics of serverless through cloud storage, clever optimizations to minimize
that can help in realizing the EPQE idea. First, thanks the amount of data exchanged, and suggest algorithms
to lightweight VM system infrastructure [
However, despite their advantages, today’s FaaS server- Our approach is to leverage an evolution of the Boxer [25]
less platforms are not adequate for general data process- system, which aims to overcome FaaS limitations (e.g.,
ing [
Apache Spark, Trino [31], etc.), (c) the configuration of the query engine instantiation, including auxiliary systems such as Zookeeper [32] (e.g., mapping of engine executors onto the resources, configuring engine settings, required storage plugins, etc.)
Our focus is on distributed data processing platforms, (SM) for the given query. If the user-supplied
specificasuch as Apache Spark or Apache Drill, instead of tradi- tions of the query engine or resources are not complete
tional database engines, such as PostgreSQL or MySQL. or are left underspecified, then (step ○ 2 ) the meta-system
We do not analyze the cost tradeofs of using AWS Lambda optimizer (MO) is used to choose all of the missing
specifor data analytics, previous works [
We report the result of using an unmodified version of Apache Drill [26] in a distributed configuration over serverless FaaS and its performance running the TPCH benchmark. This initial experiment shows that the EPQE approach is feasible and, for all but one query, executing the query with the ephemeral approach is faster than the time it takes to simply instantiate a system with Once the complete specification is determined, it is used matching configuration over AWS Fargate [ 27]/Elastic to instruct the platform provider (PP) (step ○ 3 ) to instanContainer Service (ECS) [28] (without even starting to tiate and configure the specified resources and then start run any queries). We study the start-up time of a query the configured query engine processes (and any auxilprocessing engine in this context to examine its practical iary systems). The platform provider (step ○ 4 ), using the feasibility. Finally, we also discuss preliminary results specification of initial resource allocation (a), requests that indicate that some queries run faster in one engine the resources from the underlying platform, such as net(Apache Drill) than in others (Apache Spark [29]) and worked FaaS functions, configures their networking, and vice-versa, providing initial evidence that the per-query assigns necessary names, roles, and ids to function inengine selection approach can bring important advan- stances. The query engine specification (b) determines tages. which function (or container) images are instantiated from the available catalog. Finally, before the platform 2. MetaQ Prototype provider starts the query engine, the specification of the query engine configuration (c) is used to populate the necessary configuration files and environment variables for the query engine.
Once the engine is started and ready to process queries, the session manager (step ○ 5 ) submits the user query and awaits the results from the execution engine. When the query execution completes, the session manager retrieves the results (step ○ 6 ) and returns them to the user (step ○ 7 ).
When the query execution completes, all of the resources are released, and the system scales back to zero.
We assume that the persistent data is stored in standard formats (such as Parquet, ORC, Avro, CSV) and is available through cloud storage services compatible with the common query engines (such as S3 or EBS). We restrict the set of distributed query engines considered to ones that can be used in such networked shared-disk configurations.
Our current prototype of MetaQ uses AWS Lambda FaaS functions. To run of-the-shelf query engines despite the restricted function execution environment, we utilized Boxer to provide the required but missing functionality. Boxer is a system that runs standard datacenter applications in FaaS environments, providing the expected network-of-hosts execution model. Boxer runs in every function, alongside the application processes, and
proof of concept design of the EPQE paradigm.
MetaQ has three main components: session manager (SM), platform provider (PP), and meta-system optimizer (MO). The session manager oversees end-to-end query execution and its resources, including handling communication with the client. The platform provider orchestrates the required resources and configures the environment required for the query engine execution. The meta-system optimizer is used to determine the complete specification of the resources, the query engine to be used, its conifguration, and possibly engine-specific query rewriting.
In cases when users specify the complete specification, the meta-system optimizer can be bypassed since the execution is fully specified.
Figure 1 illustrates query execution in MetaQ. To execute a query, a user (step ○ 1 ) starts MetaQ and speciifes the query and (optionally) the specifications of the query engine and resources to use for the query execution. MetaQ launches as a serverless FaaS function that can be instantiated on demand via a request to an API proxy service of a cloud provider (such as AWS API Gateway [30]).
MetaQ begins by instantiating the session manager
Startup Query Executution Fargate/ECS/EC2 min. billable Fargate/ECS Init 0.42 0.48 0 1 3 5 6 7 8
9 12 TPC-H Query 13 14 16 17 18 20 establishes an ephemeral network between the partici- reduce costs for bursty query workloads. For this study, pating functions. Boxer executes unmodified application we chose to use a variant of Boxer as our MetaQ platform processes (query engines and any auxiliary systems) in a provider (PP) component, which allowed us to instantiate FaaS environment while transparently exposing function- networked systems using AWS Lambda. For this initial to-function networking via the standard POSIX interfaces validation, we assumed that along with every considered (stream sockets etc.). To facilitate configuring the unmod- query, the user specifies the complete system specifiified distributed query engines in FaaS, Boxer is used to cation (resource allocations, query engine specification, assign roles to functions, provide name resolution, host and configuration). This bypasses the meta-engine optimembership, and coordinate query engine process execu- mizer(MO), which we plan to explore in the next stages tion. The collection of these Boxer features provides an of our research. execution environment in AWS Lambda FaaS that closely We experiment with per-query instantiations of Apache matches what is expected by distributed query engines. Drill, a general-purpose distributed SQL engine inspired
Although we show how MetaQ can run in FaaS envi- by Google Dremel [33]. We used the TPC-H benchmark ronments, its design is not tied to them. For example, to simulate the user queries to be evaluated using MetaQ. MetaQ’s components (SM,MO,PP) could execute locally Using the benchmark tools, we populated S3 cloud storon the user’s computer, and then could provide (a sub- age with data set at scale factor 10, resulting in 12 GBytes set of) standard client protocols that many distributed of data and with the largest relation with almost 60 milquery engines often expose (such as PostgreSQL stan- lion tuples. Each TPC-H query evaluation request was dard wire protocol or JDBC). Independently, there could accompanied by the complete query system specification be diferent platform providers (PP) giving access to dif- specifying (a) resources for 10 AWS Lambda functions ferent types of resources for query execution, from the with 6 vCPUs, x86_64 architecture, and 10GB of memuser’s local resources (useful for smaller workloads) to ory each, (b) Apache Drill as the query engine (the only serverless container services such as AWS Fargate or engine option in our experiment), and (c) stock configfuture serverless platforms that may provide access to uration options for Apache Drill worker nodes, a head heterogeneous hardware accelerators. node, and a single Apache Zookeeper node (required by Apache Drill).
The experiment emulates a session manager (SM) that
3. Feasibility study uses the Boxer system as the platform provided (PP) to
instantiate resources on AWS Lambda and to start Apache
3.1. Methodology Drill nodes (and Zookeeper). The experimental session
To validate the real-world feasibility of the EPQE paradigm, manager then waits for the query system to be available,
we experiment with some of the basic components of the and then submits the query and waits for the results,
MetaQ prototype design. We focus our analysis on the and returns on completion. In this study, to factor out
technical feasibility of MetaQ rather than analyzing its the efects of function caching, we ensure that only cold
cost tradeofs, [
Takeaway 1: MetaQ improves performance and re- median execution times within 10% of the slowest and duces resource usage by instantiating per-query data fastest executions. The highest observed dispersion was processing engines on FaaS infrastructure compared to for Query 20, with the slowest observed execution being containers or virtual machines. 18% slower than the median.
Figure 2 shows that a significant fraction of the query However, when we inspect the distribution of all of the execution is consumed by the startup time. The median startup times during the experiment, shown in Figure 3, time for the system to become ready to start executing we observe significantly higher variance (note that in a query is 19.67s, and (in terms of median values) that this experiment, the startup time is independent of the consumes between 30% (for query 9) and 67% (for query executed query since the client always specifies the same 6) of the total execution times for the queries we tested. complete specification with each query request, so we There are many techniques that can be used to reduce do not factor the startup times by query executed.) The this time (we have not optimized it in this experiment startup times ranged widely from 17.78s to 47.02s. In at all), from configuring the system to avoid starting particular, the measurements form two groups; of the unnecessary components to snapshotting JVM state [35, total of 70 measurements, the top 5 times (grey area in 36, 37]. Fortunately, because faster startup times are Figure 3) were above 43s, while all the remaining runs desirable for other use-cases of FaaS platforms as well, needed less than 27s to start executing the query. Our recently AWS Lambda started to ofer ability to fully initial investigation into the source of this variance indisnapshot the initial function state to avoid this issue [38]. cates that its main contributor is the time for the Apache We have not yet explored this feature, so the current Drill workers to become available after their processes results with FaaS should be treated as a conservative are started. Since the variance does not persist to the upper bound since there are further optimizations that query execution times, it suggests that these stragglers we can enable, such as restoring from snapshots. are not due to their function execution context being (per
Takeaway 2: MetaQ does not interfere with poten- manently) resource constrained. It is possible that these
tial optimizations that cloud providers could introduce functions had to fetch base images from a deeper storage
to FaaS. Its performance will only improve with these hierarchy as the worker processes were loading blocks of
optimizations, giving it an even bigger advantage over data during startup. This suggests that the meta-system
current solutions. optimizer (MO) strategies should consider the possibility
of instantiating additional workers to compensate for
this straggler phenomenon. Once enough workers are Lambada [
Takeaway 3: Although limited in scope, the experi- overhead. A lot of prior work focuses on how to mitiment demonstrates the real-world feasibility of the per- gate the data-passing limitations of FaaS infrastructure query engine paradigm. This very simple experiment by constructing custom experimental systems. leaves many possibilities for future improvements, but A first contribution and potentially the first application it already highlights the potential of our vision and mo- of the idea behind MetaQ is that it aims to run existing tivates the further exploration of the design space and platforms without having to wait until a suitable new future work on the MetaQ prototype. data processing or ML engine is developed matching the characteristics of serverless. Our approach enables run3.4. Selecting Query Engines ning complex data processing tasks at a large scale using existing mature systems, using a variety of engines taiA key aspect of the EPQE is the possibility of choos- lored to the query and data at hand, and deploying at the ing a diferent engine for each query. Although fur- scale needed while still maintaining all the advantages ther investigation is necessary, our preliminary com- of serverless. parison of query execution times of Apache Drill and Apache Spark, indicates that there likely will be a per- 4.2. Dynamically Extensible Engines formance gain from choosing diferent engines on perqueries granularity. We measured the query execution Data processing engines, such as traditional relational times of TPC-H scale factor 30 for Apache Drill and databases or many SQL-centric distributed platforms, Apache Spark using 8 AWS Lambda worker nodes. Ig- are limited along two dimensions. One is in terms of noring the startup times and only based on the relative deployment, as only one configuration is available at query execution times, we observe that for 14 of the any time. This leads to overprovisioning to make sure queries (1,4,5,6,7,9,10,11,12,13,15,16,17,22) Drill notice- the system can cope with any possible workload. The ably outperformed Spark, for 3 queries (14,19,20) Spark other is in terms of functionality. Very often, data is outperformed Drill, for 2 queries (8,18) perfromance was processed in these engines and then needs to be moved similar, while 3 queries were completed by only one of to other systems for further processing (e.g., ML training, the two engines (2,21 Spark only, 3 Drill only.) These statistical analysis, visualization). initial results suggest that, indeed, the notion of instanti- MetaQ can be used as an extension of existing engines ating a diferent engine depending on the query can be to address these two problems. In the same way we beneficial. This opens up very interesting research ques- show that one can launch a complete data processing tions in terms of how an optimizer could decide which system on serverless when a query arrives, an existing system to use. engine running on a VM could do the same to trigger additional capacity when necessary. For example, the basic mechanism presented here can be used to have Apache 4. Use Cases Drill launch additional ephemeral engines when the longrunning system is not able to cope with the additional In this section we explore use cases that could be either load. Recently, a similar approach has been explored by implemented on top of the prototype of MetaQ or would modifying an existing system, Pixels-Turbo [39] is an exrequire additional work on several aspects of the system tension of a Pixels [40] query engine that can instantiate and further research. query engines in AWS Lambda function to add elasticity to the system instantiated on long-running VMs. In the 4.1. More Eficient Data Analytics case of missing functionality for some tasks, the tranThere is a growing amount of work exploring how to best sition to another system can be done by triggering the use commercial serverless platforms for data analytics. corresponding system once the data processing engine ifnishes. This eliminates the need to have both systems running all the time and helps to automate the process rather than copying the data and transferring it manually to the other system (and then copying results back).
Complementary to these ideas is the notion of deploying a minimalist system (i.e., requiring much fewer resources) on a permanent infrastructure using VMs and then using the mechanisms of MetaQ to launch a more complete version of the system (or one tailored exactly to the task at hand) when queries arrive that require the more advanced functionality. 4.3. User-owned Data Analytics Stack Cloud providers ofer a set of Query-as-a-Service platforms, such as AWS Athena [41], which provide a simpliifed interface for large-scale analytics and charge users per byte read. However, users may still prefer to run their queries on a data analytics stack that they fully control (e.g., to optimize parameters and hardware conifgurations for their workloads). MetaQ enables users to run their own data analytics stack while still benefiting from simple abstractions and a convenient pay-per-query cost model, as resources can be acquired and released on-demand in response to load. As Palkar and Zaharia point out, users may also prefer to run their own analytics engines and web services rather than relying on out-of-the-box cloud solutions for privacy reasons [42].
This is especially true when queries involve UDFs, as these are more dificult to securely isolate in shared infrastructure deployments. By operating their own data analytics stack, users get to control how the system is configured and monitor how they are billed for the work performed for a particular task. 4.4. Data Lakes Data Lakes refer to collections of heterogeneous data that needs to be processed in a variety of diferent ways. The problem with this notion is that the processing is also highly heterogeneous, and it is the user who is responsible for handling it. Lakehouses is a new iteration of the concept that incorporates the data processing as a firstclass citizen and provides support for diferent engines, languages, etc., while automating as much as possible the task of matching data to engines and tools [43].
MetaQ is well-suited to Lakehouses as it enables dynamically selecting the engine and processing tools on the fly, and this can be done on the basis such as data types, data sizes, type of query, user requirements, or cost, etc. Furthermore, the per-query engine vision enables an intriguing possibility: sharing of auxiliary data structures across engines (indexes, partitions, zone maps, etc.) as well as creating a general infrastructure that is engine agnostic (e.g., a main memory caching layer for data to avoid having to retrieve it from slow storage every time or a results cache). Such infrastructure exists, but it is typically system specific. MetaQ opens up the possibility of seeing these aspects as orthogonal to the actual engine. In the extreme, all common modules of query engines could become serverless components dynamically added to an engine as it is instantiated with the query-specific functionality.
5.1. The Meta-Engine
to selecting the most appropriate engine for each query.
This can be done in a very simple manner by, for
instance, asking the user to specify which engine to use.
However, we are interested in automating the selection
process by building an end-to-end query system that
handles this. In a scenario where users write queries in
an engine-agnostic syntax (for example, in a declarative
language such as SQL), MetaQ’s meta-system optimizer
could inspect the query and determine which engine is
the most eficient given the data types, its type (static
or streaming), the type of operations required, etc. This
leads to cross-engine optimizations, such as picking the
engine that is faster to perform a given operation
provided by several engines. The main research question
is how to derive meta-system optimizer policies. One
possible approach is to extend the domain of automatic
configuration systems [
per query, it is now possible to optimize the deployment where the engine will run for every query. Such deployment configuration could determine the amount of resources used, such as the CPU and/or memory budget.
Such configuration could be inferred by analyzing the query and data inputs to estimate the amount of data that would be processed and, therefore, the amount of compute and memory necessary to finish the query within a particular time frame. From another perspective, it is now possible to dynamically find tradeofs between execution time and price for each query. This tradeof could also be exposed to users as a way to prioritize interactive services provided by the cloud. This results in inefiqueries over batch workloads. ciencies that are dificult to address: over-provisioning, coarse resource allocation, generic engine configurations, 5.3. Query Scheduling and Caching low utilization, etc. In this paper, we put forward the idea of ephemeral per-query engines: selected query engines Beyond automatically sizing and optimizing per-query dynamically instantiated when a query arrives and redeployments, it is also possible to schedule query execu- moved when it terminates. In the paper, we have outlined tion on nodes that have some locally cached data or that the idea, discussed its potential to address many of the are close to storage nodes. For example, if a workload re- limitations of current deployments, provided a feasibility quires two queries to be executed, the second query could study, and demonstrated that, while there is still much be scheduled for execution on the same physical node(s) work to do, it is possible to implement it in current FaaS that was used to execute the previous one. To keep data platforms. The initial experiments are highly encouraglocal, caching approaches such as Faa$t cache [44] can ing. They show that existing engines can be suficiently also be used to keep the output of queries. quickly instantiated on demand to run a single query. Building on this basis, we have also discussed and pre5.4. System Infrastructure sented several research directions that can be pursued based on the ideas and results presented here.
totype uses Boxer as its platform provider. Boxer (and therefore MetaQ) do not require any cloud provider intervention and can be deployed today in AWS Lambda. However, Boxer is not yet feature complete in terms of interfaces, networking support, reliability, and integration within larger systems. That is something that we are working on at the moment so as to have a more solid basis for the system. Similarly, Boxer was initially built for AWS Lambda. We are in the process of studying how to port Boxer to other commercial serverless oferings. Doing so would open yet another wave of exciting opportunities, like triggering serverless jobs across heterogeneous clouds using the networking capabilities available in Boxer. 5.5. Generalizing to Other Engines Our experiments are only a first step towards the perquery engine vision. We plan to test this paradigm and our MetaQ prototype on a wider range of data processing engines and platforms on top of the existing prototype to make sure it can indeed be used as a general-purpose distributed computing platform equivalent to what can be done on a VM. Systems that we are in the process of testing include Apache Spark, Trino, Databend [45], Flink [46], Clickhouse [47]. Having them running on the same serverless platform will also ofer a great opportunity to study the engine designs that are most suitable for serverless, providing very valuable information on the road toward serverless native engines.
[38] Improving startup performance with Lambda Snap- house: A new generation of open platforms that Start, 2023. URL: https://docs.aws.amazon.com/ unify data warehousing and advanced analytics, in: lambda/latest/dg/snapstart.html, (accessed: 2023- 11th Conference on Innovative Data Systems Re03-01). search, CIDR 2021, Virtual Event, January 11-15, [39] H. Bian, T. Sha, A. Ailamaki, Using cloud functions 2021, Online Proceedings, www.cidrdb.org, 2021.
as accelerator for elastic data analytics 1 (2023). [44] F. Romero, G. I. Chaudhry, I. n. Goiri, P. Gopa, P. Ba[40] H. Bian, A. Ailamaki, Pixels: An eficient column tum, N. J. Yadwadkar, R. Fonseca, C. Kozyrakis, store for cloud data lakes, in: 2022 IEEE 38th Inter- R. Bianchini, Faa$t: A transparent auto-scaling national Conference on Data Engineering (ICDE), cache for serverless applications, Association for 2022, pp. 3078–3090. Computing Machinery, New York, NY, USA, 2021. [41] Amazon Athena, 2020. URL: http://docs.aws. [45] Databend, 2023. URL: https://databend.rs/, (acamazon.com/athena/, (accessed: 2020-08-17). cessed: 2023-06-20). [42] S. Palkar, M. Zaharia, Diy hosting for online privacy, [46] Apache Flink, 2023. URL: https://flink.apache.org/, in: Proceedings of the 16th ACM Workshop on Hot (accessed: 2023-03-01).
Topics in Networks, HotNets-XVI, 2017. [47] ClickHouse, 2023. URL: https://clickhouse.com/, (ac[43] M. Zaharia, A. Ghodsi, R. Xin, M. Armbrust, Lake- cessed: 2023-06-20).