Short query prioritization for data processing service
A cluster computing system maintains a first set of queues for short queries and a set second set for longer queries. The first set is allocated a majority of the cluster's processing resources and processes queries on a first in first out basis. The second set is allocated a minority of the cluster's processing resources which are shared among queries in the second set. Accordingly, the system assigns each query to the first set of queues for a fixed amount of resource time. While a query is processing, the system monitors the query's resource time and reassigns the query to the second set of queues if the query has not completed within the allotted amount of resource time. Thus, short queries receive the necessary resources to complete quickly without getting stuck behind longer queries while ensuring that longer queries continue to make progress.
1 . A method, comprising:
accessing a cluster including a driver node and one or more executor nodes each including one or more task engines, the cluster configured to schedule queries according to at least a first queue, a second queue, and a third queue, the first queue storing information on queries with a higher priority than queries assigned to the second queue, and the second queue storing information on queries with a higher priority than queries assigned to the third queue;
receiving a query and assigning the query to the first queue, the query divided into a set of tasks;
allocating a number of task engines in a first set of executor nodes to process the set of tasks of the query;
monitoring a resource time for the query while the query is being executed at the first set of executor nodes;
responsive to determining that the resource time for the query is equal to or above a threshold resource time, assigning the query to the second queue;
responsive to assigning the query to the second queue, executing the tasks of the query on a second set of executor nodes;
responsive to determining that a second resource time for the query is equal to or above a second threshold resource time, assigning the query to the third queue; and
responsive to assigning the query to the third queue, executing the tasks of the query on a third set of executor nodes.
2 . The method of claim 1 , wherein a number of executor nodes in the third set of executor nodes is smaller than a number of executor nodes in the first set of executor nodes.
3 . The method of claim 1 , further comprising:
receiving another query for execution;
responsive to assigning the query to the second queue, assigning task engines to the another query assigned to the first queue; and
reallocating at least a portion of executor nodes from the first set of executor nodes to process the other query assigned to the first queue.
4 . The method of claim 1 , wherein monitoring the resource time comprises:
determining, for each task of the query, an execution time to execute the task on a respective task engine; and
summing the execution time of the tasks of the query to determine the resource time.
5 . The method of claim 1 , further comprising:
monitoring a number of tasks for queries assigned to the first queue;
determining a number of task engines to be reserved for queries assigned to the first queue based on the monitored number of tasks; and
reserving the number of task engines on the cluster for queries assigned to the first queue for a next time step.
6 . The method of claim 5 , wherein the number of tasks for queries assigned to the first queue includes pending and running tasks, and wherein reserving the number of task engines on the cluster for queries assigned to the first queue for the next time step includes:
applying an exponential moving average (EMA) function on a sum of the pending and the running tasks; and
reserving Y task engines for the queries in the first queue, where Y=f (aggressiveness factor, X), where X is the sum of the pending and the running tasks and the aggressiveness factor is a constant.
7 . The method of claim 1 , further comprising:
monitoring a number of task engines for processing queries assigned to the third queue;
responsive to determining that the number of task engines is equal to or above a threshold, reducing at least the third set of executor nodes dedicated to processing the query.
8 . A non-transitory computer-readable storage medium comprising stored instructions that, when executed by one or more computer processors, causes the one or more computer processors to:
access a cluster including a driver node and one or more executor nodes each including one or more task engines, the cluster configured to schedule queries according to at least a first queue, a second queue, and a third queue, the first queue storing information on queries with a higher priority than queries assigned to the second queue, and the second queue storing information on queries with a higher priority than queries assigned to the third queue;
receive a query and assigning the query to the first queue, the query divided into a set of tasks;
allocate a number of task engines in a first set of executor nodes to process the set of tasks of the query;
monitor a resource time for the query while the query is being executed at the first set of executor nodes;
responsive to determining that the resource time for the query is equal to or above a threshold resource time, assign the query to the second queue;
responsive to assigning the query to the second queue, execute the tasks of the query on a second set of executor nodes;
responsive to determining that a second resource time for the query is equal to or above a second threshold resource time, assign the query to the third queue; and
responsive to assigning the query to the third queue, execute the tasks of the query on a third set of executor nodes.
9 . The non-transitory computer-readable storage medium of claim 8 , wherein a number of executor nodes in the third set of executor nodes is smaller than a number of executor nodes in the first set of executor nodes.
10 . The non-transitory computer-readable storage medium of claim 8 , wherein the instructions further cause the one or more computer processors to:
receive another query for execution;
responsive to assigning the query to the second queue, assign task engines to the another query assigned to the first queue; and
reallocate at least a portion of executor nodes from the first set of executor nodes to process the other query assigned to the first queue.
11 . The non-transitory computer-readable storage medium of claim 8 , wherein the instructions further cause the one or more computer processors to:
determine, for each task of the query, an execution time to execute the task on a respective task engine; and
sum the execution time of the tasks of the query to determine the resource time.
12 . The non-transitory computer-readable storage medium of claim 8 , wherein the instructions further cause the one or more computer processors to:
monitor a number of tasks for queries assigned to the first queue;
determine a number of task engines to be reserved for queries assigned to the first queue based on the monitored number of tasks; and
reserve the number of task engines on the cluster for queries assigned to the first queue for a next time step.
13 . The non-transitory computer-readable storage medium of claim 12 , wherein the number of tasks for queries assigned to the first queue includes pending and running tasks, and wherein the instructions further cause the one or more computer processors to:
apply an exponential moving average (EMA) function on a sum of the pending and the running tasks; and
reserve Y task engines for the queries in the first queue, where Y=f(aggressiveness factor, X), where X is the sum of the pending and the running tasks and the aggressiveness factor is a constant.
14 . The non-transitory computer-readable storage medium of claim 8 , wherein the instructions further cause the one or more computer processors to:
monitor a number of task engines for processing queries assigned to the third queue;
responsive to determining that the number of task engines is equal to or above a threshold, reduce at least the third set of executor nodes dedicated to processing the query.
15 . A computer system comprising:
one or more computer processors; and
one or more non-transitory computer readable storage mediums comprising stored instructions that, when executed by one or more computer processors, causes the one or more computer processors to:
access a cluster including a driver node and one or more executor nodes each including one or more task engines, the cluster configured to schedule queries according to at least a first queue, a second queue, and a third queue, the first queue storing information on queries with a higher priority than queries assigned to the second queue, and the second queue storing information on queries with a higher priority than queries assigned to the third queue;
receive a query and assigning the query to the first queue, the query divided into a set of tasks;
allocate a number of task engines in a first set of executor nodes to process the set of tasks of the query;
monitor a resource time for the query while the query is being executed at the first set of executor nodes;
responsive to determining that the resource time for the query is equal to or above a threshold resource time, assign the query to the second queue;
responsive to assigning the query to the second queue, execute the tasks of the query on a second set of executor nodes;
responsive to determining that a second resource time for the query is equal to or above a second threshold resource time, assign the query to the third queue; and
responsive to assigning the query to the third queue, execute the tasks of the query on a third set of executor nodes.
16 . The computer system of claim 15 , wherein a number of executor nodes in the third set of executor nodes is smaller than a number of executor nodes in the first set of executor nodes.
17 . The computer system of claim 15 , wherein the instructions further cause the one or more computer processors to:
receive another query for execution;
responsive to assigning the query to the second queue, assign task engines to the another query assigned to the first queue; and
reallocate at least a portion of executor nodes from the first set of executor nodes to process the other query assigned to the first queue.
18 . The computer system of claim 15 , wherein the instructions further cause the one or more computer processors to:
determine, for each task of the query, an execution time to execute the task on a respective task engine; and
sum the execution time of the tasks of the query to determine the resource time.
19 . The computer system of claim 15 , wherein the instructions further cause the one or more computer processors to:
monitor a number of tasks for queries assigned to the first queue;
determine a number of task engines to be reserved for queries assigned to the first queue based on the monitored number of tasks; and
reserve the number of task engines on the cluster for queries assigned to the first queue for a next time step.
20 . The computer system of claim 19 , wherein the number of tasks for queries assigned to the first queue includes pending and running tasks, and wherein the instructions further cause the one or more computer processors to:
apply an exponential moving average (EMA) function on a sum of the pending and the running tasks; and
reserve Y task engines for the queries in the first queue, where Y=f (aggressiveness factor, X), where X is the sum of the pending and the running tasks and the aggressiveness factor is a constant.