Dynamic tree determination for data processing

• US 8,312,037 B1
• Filed: 08/28/2008
• Issued: 11/13/2012
• Est. Priority Date: 08/28/2008
• Status: Active Grant

• Current Assignee: Amazon Technologies, Inc. (Amazon.com, Inc.)
• Original Assignee: Amazon Technologies, Inc. (Amazon.com, Inc.)
• Inventors: Bacthavachalu, Govindaswamy, Gavares, Peter Grant, Badran, Ahmed A., Scharf, James E. Jr.
• Primary Examiner(s): Peng, Huawen
• Application Number:
  US12/200,821
• Time in Patent Office: 1,538 Days
• Field of Search: 707/770, 707/773, 707/802, 707/769, 707/791, 712/30, 712/11
• US Class Current: 707/769
• CPC Class Codes: G06F 16/22 Indexing; Data structures therefor; Storage structures
• CPC Class Codes: G06F 16/2246 Trees, e.g. B+trees
• CPC Class Codes: G06F 16/24532 of parallel queries
• CPC Class Codes: G06F 16/2455 Query execution
• CPC Class Codes: G06F 16/24556 Aggregation; Duplicate elimination
• CPC Class Codes: G06F 16/2471 Distributed queries
• CPC Class Codes: G06F 16/275 Synchronous replication
• CPC Class Codes: G06F 9/4881 Scheduling strategies for dispatcher, e.g. round robin, multi-level priority queues
• CPC Class Codes: G06F 9/5066 Algorithms for mapping a plurality of inter-dependent sub-tasks onto a plurality of physical CPUs mappping at compile time, see G06F8/451
• CPC Class Codes: G06N 5/02 Knowledge representation; Symbolic representation

Abstract

Data can be processed in parallel across a cluster of nodes using a parallel processing framework. Using Web services calls between components allows the number of nodes to be scaled as necessary, and allows developers to build applications on the framework using a Web services interface. A job scheduler works together with a queuing service to distribute jobs to nodes as the nodes have capacity, such that jobs can be performed in parallel as quickly as the nodes are able to process the jobs. Data can be loaded efficiently across the cluster, and levels of nodes can be determined dynamically to process queries and other requests on the system.

Claims

1. A method of determining a node hierarchy for processing a set of queries, comprising:

receiving an instance query and a summary query to process data using a cluster of nodes;

specifying a first set of nodes in the cluster to process the instance query, the number of nodes to be included in the first set of nodes based at least in part upon a predicted number of first results and an expected capacity of each node of the first set of nodes;

determining at least a second set of nodes in the cluster based at least in part upon the predicted number of first results for the instance query to process the summary query, each node of the second set of nodes being a parent node to at least one child node in the first set of nodes and the second set of nodes being distinct from the first set of nodes;

scheduling a plurality of jobs each corresponding to at least one of the instance or summary queries, each job to be executed on a single node;

causing each node of the first set of nodes to process the instance query, at least a portion of the first set of nodes operable to process the instance query in parallel;

causing at least a portion of the second set of nodes to process the summary query when a respective portion of the first set of nodes finishes processing the instance query, at least a portion of the second set of nodes operable to process the summary query in parallel; and

storing a result of the summary query to a specified location,

wherein the second set of nodes is operable to have a plurality of hierarchical levels such that a parent node of the second set of nodes processes results of at least two child nodes in the second set of nodes, the number of hierarchical levels being based at least in part upon the predicted number of first results and an expected capacity of each node of the second set of nodes.

2. A method according to claim 1, further comprising:

determining a third set of nodes to process an intermediate query, each node of the third set of nodes being a parent node to at least one child node of the first set of nodes and a child node to a parent node in the second set of nodes; and

causing each node of the third set of nodes to process the intermediate query using results from the instance query on each child node,

wherein each node of the second set of nodes is able to further process the summary query using results from the intermediate query on each respective child node.

3. A method according to claim 1, further comprising:

tracking dependencies between jobs such that jobs for the second set of nodes are not processed until the corresponding jobs the first set of nodes are processed.

4. A method of determining a node hierarchy for processing a request, comprising:

specifying a first set of nodes in a cluster to process a first portion of a request, the number of nodes in the first set of nodes determined based at least in part upon a predicted number of first results and an expected capacity of each node of the first set of nodes;

determining a second set of nodes in the cluster based at least in part upon the predicted number of first results for the first portion of the request to process a second portion of the request, each node of the second set of nodes being a parent node to at least one child node in the first set of nodes and the second set of nodes being distinct from the first set of nodes;

causing each node of the first set of nodes to process the first portion of the request, at least a portion of the first set of nodes operable to process the first portion in parallel;

causing at least a portion of the second set of nodes to process the second portion of the request when a respective portion of the first set of nodes finishes processing the first portion of the request, at least a portion of the second set of nodes operable to process the second portion in parallel; and

wherein the second set of nodes is operable to have a plurality of hierarchical levels such that a parent node of the second set of nodes processes results of at least two child nodes in the second set of nodes, the number of hierarchical levels being based at least in part upon the predicted number of first results and an expected capacity of each node of the second set of nodes.

5. A method according to claim 4, wherein the request is processed in a database system, further comprising:

automatically determining the predicted number of first results for the first portion of the request based on historical information for the system.

6. A method according to claim 4, further comprising:

receiving a number of nodes to be included in the first set of nodes as specified by a user.

7. A method according to claim 4, further comprising:

determining a third set of nodes to process a third portion of the request, each node of the third set of nodes being a parent node to at least one child node of the first set of nodes and a child node to a parent node in the second set of nodes; and

causing each node of the third set of nodes to process the third portion using results from the first portion on each child node,

wherein each node of the second set of nodes is able to further process the second portion using results from the third portion on each child node.

8. A method according to claim 7, wherein:

the cluster of nodes communicates using Web services and is scalable to add additional nodes.

9. A method according to claim 4, wherein:

each node comprises at least one of a computing device, an instance on a computing device, or a virtual server.

11. A method according to claim 4, further comprising:

passing results from each node of the first set of nodes to a corresponding node of the second set of nodes using a file transfer service.

12. A method according to claim 4, further comprising:

enabling a user to upload a library to each node to be used in processing at least the first portion of the request.

13. A method according to claim 4, wherein:

the second set of nodes is determined dynamically further based upon at least a current capacity of each node at the time for processing the summary query.

14. A method according to claim 4, further comprising:

placing a reference for each set of results in a queue for use by a parent node.

15. A method according to claim 4, wherein:

the number of nodes and levels in the second set of nodes is determined using at least one of a tree determination algorithm or a merge ratio.

16. A method according to claim 4, further comprising:

creating a temporary table on each node for storing results of the respective job, each temporary table being formed according to a specified schema.

17. A method according to claim 4, wherein:

at least a portion of the second portion of the request combines results from the first portion of the request.

18. A method according to claim 4, further comprising:

receiving a selection of the specified location from a user.

19. A method according to claim 4, further comprising:

enabling a user to submit a request to further process a result previously stored to the specified location.

20. A method according to claim 4, further comprising:

duplicating a table of data across each node in the cluster.

10. A method of determining a node hierarchy for processing a request, comprising:

specifying a first set of nodes in a cluster to process a first portion of a request, the number of nodes in the first set of nodes determined based at least in part upon a predicted number of first results and an expected capacity of each node of the first set of nodes;

determining a second set of nodes in the cluster based at least in part upon the predicted number of first results for the first portion of the request to process a second portion of the request, each node of the second set of nodes being a parent node to at least one child node in the first set of nodes and the second set of nodes being distinct from the first set of nodes;

causing each node of the first set of nodes to process the first portion of the request, at least a portion of the first set of nodes operable to process the first portion in parallel;

causing at least a portion of the second set of nodes to process the second portion of the request when a respective portion of the first set of nodes finishes processing the first portion of the request, at least a portion of the second set of nodes operable to process the second portion in parallel;

wherein the second set of nodes is operable to have a plurality of hierarchical levels such that a parent node of the second set of nodes processes results of at least two child nodes in the second set of nodes, the number of hierarchical levels being based at least in part upon the predicted number of first results and an expected capacity of each node of the second set of nodes; and

wherein causing each node of the first set of nodes to process the first portion of the request and causing each of the second set of nodes to process the second portion of the request collectively comprises;

scheduling a plurality of jobs each corresponding to at least one of the first and or second portions of the request, each job to be executed on a single node;

tracking dependencies between jobs such that jobs for the second set of nodes are not processed until the corresponding jobs for the first set of nodes are processed.

21. A system for determining a node hierarchy for processing a request, comprising:

a processor; and

a memory device including instructions that, when executed by the processor, cause the processor to;

specify a first set of nodes in a cluster to process a first portion of a request, the number of nodes in the first set of nodes determined based at least in part upon a predicted number of first results and an expected capacity of each node of the first set of nodes;

determine a second set of nodes in the cluster based at least in part upon the predicted number of first results for the first portion of the request to process a second portion of the request, each node of the second set of nodes being a parent node to at least one child node in the first set of nodes and the second set of nodes being distinct from the first set of nodes;

cause each node of the first set of nodes to process the first portion of the request, at least a portion of the first set of nodes operable to process the first portion in parallel;

cause at least a portion of the second set of nodes to process the second portion of the request when a respective portion of the first set of nodes finishes processing the first portion of the request, at least a portion of the second set of nodes operable to process the second portion in parallel; and

wherein the second set of nodes is operable to have a plurality of hierarchical levels such that a parent node of the second set of nodes processes results of at least two child nodes in the second set of nodes, the number of hierarchical levels being based at least in part upon the predicted number of first results and an expected capacity of each node of the second set of nodes.

22. A system according to claim 21, wherein the request is processed in a database system, wherein the memory device further includes instructions that, when executed by the processor, cause the processor to:

automatically determine the predicted number of first results for the first portion of the request based on historical information for the system.

23. A system according to claim 21, wherein the memory device further includes instructions that, when executed by the processor, cause the processor to:

receive a number of nodes to be included in the first set of nodes as specified by a user.

24. A computer program product embedded in a non-transitory computer readable storage medium having program instructions stored thereon that upon execution by a processor causes a computer system to determine a node hierarchy for processing a request, the computer program product comprising:

program instructions for specifying a first set of nodes in a cluster to process a first portion of a request, the number of nodes in the first set of nodes determined based at least in part upon a predicted number of first results and an expected capacity of each node of the first set of nodes;

program instructions for determining a second set of nodes in the cluster based at least in part upon a predicted number of first results for the first portion of the request to process a second portion of the request, each node of the second set of nodes being a parent node to at least one child node in the first set of nodes and the second set of nodes being distinct from the first set of nodes;

program instructions for causing each node of the first set of nodes to process the first portion of the request, at least a portion of the first set of nodes operable to process the first portion in parallel;

program instructions for causing at least a portion of the second set of nodes to process the second portion of the request when a respective portion of the first set of nodes finishes processing the first portion of the request, at least a portion of the second set of nodes operable to process the second portion in parallel; and

wherein the second set of nodes is operable to have a plurality of hierarchical levels such that a parent node of the second set of nodes processes results of at least two child nodes in the second set of nodes, the number of hierarchical levels being based at least in part upon the predicted number of first results and an expected capacity of each node of the second set of nodes.

25. A computer program product according to claim 24, wherein the request is processed in a database system, further comprising:

program instructions for automatically determining the predicted number of first results for the first portion of the request based on historical information for the system.