Dynamic scaling of a cluster of computing nodes used for distributed execution of a program

• US 8,260,840 B1
• Filed: 06/28/2010
• Issued: 09/04/2012
• Est. Priority Date: 06/28/2010
• Status: Active Grant

• Current Assignee: Amazon Technologies, Inc. (Amazon.com, Inc.)
• Original Assignee: Amazon Technologies, Inc. (Amazon.com, Inc.)
• Inventors: Sirota, Peter, Cole, Richard J., Aggarwal, Vaibhav, Hitchcock, Andrew J., Vasanth, Jai
• Primary Examiner(s): Jean Gilles, Jude
• Application Number:
  US12/825,073
• Time in Patent Office: 799 Days
• Field of Search: 709/201
• US Class Current: 709/201
• CPC Class Codes: G06F 9/5061 Partitioning or combining of resources

Abstract

Techniques are described for managing distributed execution of programs, including by dynamically scaling a cluster of multiple computing nodes used to perform ongoing distributed execution of a program, such as to increase and/or decrease the quantity of computing nodes in the cluster at various times and for various reasons. An architecture may be used that facilitates the dynamic scaling of a cluster, including by having at least some of the computing nodes act as core nodes that each participate in a distributed storage system for the distributed program execution, and having one or more other computing nodes that act as auxiliary nodes that do not participate in the distributed storage system. If computing nodes are selected to be removed from the cluster during ongoing distributed execution of a program, one or more nodes of the auxiliary computing node type may be selected for the removal.

Claims

1. A computer-implemented method comprising:

receiving configuration information regarding execution of an indicated program in a distributed manner that includes executing a plurality of jobs of the indicated program, the configuration information being received by one or more computing systems configured to provide a distributed program execution service that performs distributed execution of programs for clients using a plurality of configurable computing nodes;

automatically determining a subset of multiple of the plurality of configurable computing nodes to be used for the execution of the indicated program in accordance with the received configuration information, the multiple computing nodes of the determined subset including a first group of one or more computing nodes to act as core computing nodes that each participate in a distributed storage system storing information used in the execution of the indicated program, the multiple computing nodes of the determined subset further including a second group of one or more computing nodes to act as auxiliary computing nodes that do not participate in the distributed storage system, the automatic determining of the subset being performed by the one or more configured computing systems;

initiating the execution of the indicated program in the distributed manner on the multiple computing nodes at a first time by executing one or more of the plurality of jobs of the indicated program on each of the multiple computing nodes, the initiating of the execution of the indicated program being performed by the one or more configured computing systems; and

at a second time subsequent to the first time and during the execution of the indicated program, determining to reduce a quantity of the multiple computing nodes that are used for ongoing execution of the indicated program, and automatically reducing the quantity of the multiple computing nodes by initiating removal of at least one of the auxiliary computing nodes of the second group without removing any of the core computing nodes of the first group, the automatic reducing of the quantity of the multiple computing nodes being performed by the one or more configured computing systems.

2. The method of claim 1 wherein the determined subset of multiple computing nodes are part of a cluster that is created for use in the execution of the indicated program on behalf of a first client from whom the configuration information is received, and wherein the determining to reduce the quantity of the multiple computing nodes is performed based on one of an instruction received at the second time from the first client and of an automated determination at the second time by the one or more configured computing systems that one or more scaling criteria defined by the first client prior to the second time for the cluster are satisfied at the second time.

3. The method of claim 2 wherein the determining to reduce the quantity of the multiple computing nodes is performed based on the automated determination at the second time by the one or more configured computing systems that the one or more defined scaling criteria are satisfied at the second time, and wherein the one or more defined scaling criteria are satisfied at the second time based at least in part on a stage of the execution of the indicated program that is specified by at least one of the one or more defined scaling criteria.

4. The method of claim 2 wherein the determining to reduce the quantity of the multiple computing nodes is performed based on the automated determination at the second time by the one or more configured computing systems that the one or more defined scaling criteria are satisfied at the second time, and wherein the one or more defined scaling criteria are satisfied at the second time based at least in part on a monitored state of use of computing resources by the execution of the indicated program that is specified by at least one of the one or more defined scaling criteria.

5. The method of claim 1 wherein the method further comprises, at a third time subsequent to the first time and during the execution of the indicated program, automatically modifying at the third time the quantity of the multiple computing nodes that are used for the ongoing execution of the indicated program, the automatic modifying being performed by the one or more configured computing systems and including at least one of automatically removing at least one of the auxiliary computing nodes of the second group, automatically adding at least one computing node to the auxiliary computing nodes of the second group, automatically removing at least one of the core computing nodes of the first group, and automatically adding at least one computing node to the core computing nodes of the first group.

6. The method of claim 5 wherein the plurality of configurable computing nodes include computing nodes of multiple distinct resource configuration types that each have a distinct configuration of one or more computing resources, and wherein the automatically modifying at the third time of the quantity of the multiple computing nodes includes selecting one of the multiple distinct resource configuration types and performing one or more modifications involving one or more computing nodes of the selected one resource configuration type.

7. The method of claim 1 wherein the distributed storage system is a distributed file system, wherein each of core computing nodes stores a portion of the distributed file system on local storage of that core computing node, wherein a first of the core computing nodes stores a first and second subset of input data for the indicated program in the portion of the distributed file system on the first core computing node, wherein the executing of the one or more jobs of the indicated program on the first core computing node includes selecting a first execution job for the first core computing node based on the first execution job using the first subset of input data that is stored by the first core computing node and includes selecting a second execution job that uses the second subset of input data for a first of the auxiliary computing nodes, and wherein the executing of the second execution job by the first auxiliary computing node includes obtaining the second subset of input data from the first core computing node.

8. The method of claim 7 wherein the determined subset of multiple computing nodes include computing nodes at multiple distinct geographical locations, wherein the first core computing node and the first auxiliary computing node are co-located at a first of the multiple geographical locations, and wherein the selecting of the first auxiliary computing node for the second execution job that uses the second subset of input data is based at least in part of the first auxiliary computing node being co-located with the first core computing node that stores the second subset of input data.

9. The method of claim 1 wherein the determined subset of multiple computing nodes are part of a cluster that is created for use in the execution of the indicated program on behalf of a first client, and wherein the method further comprises monitoring ongoing execution of the indicated program on the cluster to generate status information related to the ongoing execution, the generated status information including first status information for the first group of computing nodes acting as core computing nodes and including second status information for the second group of computing nodes acting as auxiliary computing nodes, and providing at least some of the generated first status information and generated second status information to the first client during the ongoing execution of the indicated program to enable the first client to track the ongoing execution with respect to the first and second groups of computing nodes.

10. The method of claim 1 wherein the indicated program is configured to perform one or more map functions on each of multiple input data subsets and to perform one or more reduce functions on results of the one or more map functions, wherein the method further comprises generating at least some of the plurality of jobs to each implement at least one of the map functions or at least one of the reduce functions and further comprises completing the execution of the plurality of jobs of the indicated program and further comprises providing final results from the execution to a first client who supplied the received configuration information, wherein the distributed program execution service provides the plurality of configurable computing nodes and uses virtualization technology such that the plurality of computing nodes include, for each of multiple physical computing systems, multiple virtual machines hosted by the physical computing system that are each able to execute at least one portion of a program, and wherein the multiple computing nodes on which the indicated program is executed are each a hosted virtual machine.

11. A non-transitory computer-readable medium whose contents configure a computing system to perform a method, the method comprising:

initiating execution of an indicated program in a distributed manner on a cluster of multiple computing nodes at a first time;

at a second time subsequent to the first time and while the execution of the indicated program is ongoing, determining, by the configured computing system, to modify a quantity of the multiple computing nodes in the cluster during the ongoing execution of the indicated program;

identifying, by the configured computing system, a first subset of the multiple computing nodes that each are part of distributed storage at the second time for the ongoing execution of the indicated program, and a second subset of one or more of the multiple computing nodes that are distinct from the multiple computing nodes of the first subset and that are not part of the distributed storage at the second time; and

initiating, by the configured computing system, a change in the quantity of the multiple computing nodes of the cluster while the execution of the indicated program is ongoing in response to the determining to modify the quantity, the initiated change in the quantity being performed for one of the first subset and of the second subset that is selected for the initiated change based at least in part on a type of the determined modifying of the quantity, the initiating of the change being performed by the configured computing system.

12. The non-transitory computer-readable medium of claim 11 wherein the configured computing system is part of a distributed program execution service that provides a plurality of computing nodes configurable to perform distributed execution of programs for clients, wherein the initiating of the execution of the indicated program is performed for a first client and includes attempting to execute one or more jobs of the indicated program on the multiple computing nodes using indicated input data, wherein the multiple computing nodes of the cluster are a subset of the plurality of computing nodes that are provided for use on behalf of the first client, wherein the computing nodes of the first subset that are part of the distributed storage at the second time include a first group of one or more core computing nodes that are each part of a distributed storage system for the first client that stores the indicated input data, wherein the computing nodes of the second subset are a second group of one or more auxiliary computing nodes that are not part of the distributed storage system for the first client, and wherein the initiating of the change in the quantity of the multiple computing nodes includes performing a selection of the one of the first subset and of the second subset for the initiated change based at least in part on whether the initiated change is to remove one or more computing nodes from the cluster or is to add one or more computing nodes to the cluster.

13. The non-transitory computer-readable medium of claim 12 wherein the type of the determined modifying of the quantity of the multiple computing nodes in the cluster is to reduce the quantity of the multiple computing nodes in the cluster by at least one, wherein the performing of the selection of the one of the first subset and of the second subset includes selecting the auxiliary computing nodes of the second subset based on the reducing of the quantity, and wherein the initiating of the change in the quantity includes removing one or more of the auxiliary computing nodes from the cluster.

14. The non-transitory computer-readable medium of claim 11 wherein the type of the determined modifying of the quantity of the multiple computing nodes in the cluster includes one of reducing the quantity of the multiple computing nodes in the cluster and of increasing the quantity of the multiple computing nodes in the cluster, and wherein the initiating of the change in the quantity of the multiple computing nodes includes performing a selection of the first subset for the initiated change and includes modifying a quantity of computing nodes of the first subset.

15. The non-transitory computer-readable medium of claim 11 wherein the computer-readable medium is a memory of the configured computing system that stores the contents, and wherein the contents are instructions that when executed program the computing system to perform the method.

16. A computing system, comprising:

one or more processors; and

one or more modules of a distributed execution service that are configured to, when executed by at least one of the one or more processors, manage distributed execution of programs for clients by, for each of one or more of the clients;

receiving information from the client regarding execution of an indicated program in a distributed manner on a cluster of multiple computing nodes;

initiating the execution of the indicated program in the distributed manner on the cluster of multiple computing nodes at a first time;

at a second time subsequent to the first time and while the execution of the indicated program is ongoing, determining to modify a quantity of first computing nodes in the cluster that are not part of distributed storage for information used for the ongoing execution of the indicated program, the first computing nodes being distinct from at least some computing nodes in the cluster at the second time that each are part of the distributed storage; and

initiating a change in the quantity of the first computing nodes of the cluster while the execution of the indicated program is ongoing and in response to the determining to modify the quantity.

17. The computing system of claim 16 wherein the distributed execution service is a distributed program execution service that provides a plurality of computing nodes configurable to perform distributed execution of programs for the clients of the distributed program execution service, wherein the multiple computing nodes of the cluster for a first of the one or more clients are a subset of the plurality of computing nodes, wherein the at least some computing nodes in the cluster of the first client at the second time that are part of the distributed storage for the first client includes a first group of one or more core computing nodes that are each part of a distributed storage system for the first client, wherein the first computing nodes in the cluster of the first client on which the change in the quantity is initiated are a second group of one or more auxiliary computing nodes that are not part of the distributed storage system and that each executes one or more of a plurality of jobs of the indicated program, and wherein the one or more computing nodes of the cluster for the first client on which one or more jobs are attempted to be executed at the first time include the one or more auxiliary computing nodes.

18. The computing system of claim 17 wherein the information received from the first client is configuration information for the execution of the indicated program for the first client that includes an indication of a quantity of the multiple computing nodes in the cluster for the first client at the first time and that includes one or more criteria that is used in the automatic determining at the second time to modify the quantity of the first computing nodes in the cluster for the first client, and wherein the one or more computing nodes of the cluster for the first client on which one or more jobs are attempted to be executed at the first time further include the one or more core computing nodes.

19. The computing system of claim 16 wherein, for a first of the one or more clients, the cluster for the first client includes one or more of the first computing nodes at the second time, and wherein the automatically initiated change in the quantity of the first computing nodes in the cluster for the first client is to remove at least one of the one or more first computing nodes while the execution of the indicated program for the first client is ongoing on one or more other computing nodes of the cluster for the first client.

20. The computing system of claim 16 wherein, for a first of the one or more clients, the cluster for the first client includes one or more of the first computing nodes at the second time, and wherein the automatically initiated change in the quantity of the first computing nodes in the cluster for the first client is to add at least one additional first computing node while the execution of the indicated program for the first client is ongoing on other of the first computing nodes of the cluster for the first client.

21. The computing system of claim 16 wherein, for a first of the one or more clients, the cluster for the first client includes a quantity of zero of the first computing nodes at the second time, and wherein the automatically initiated change in the quantity of the first computing nodes in the cluster for the first client is to add at least one first computing node while the execution of the indicated program for the first client is ongoing on other computing nodes of the cluster for the first client.

22. The computing system of claim 16 wherein the one or more modules are further configured to, for a first of the one or more clients, automatically initiate a change in quantity of the at least some computing nodes in the cluster of the first client at the second time that each are part of the distributed storage for the first client.

23. The computing system of claim 16 wherein the one or more modules each include software instructions for execution by the computing system.

24. The computing system of claim 16 wherein the one or more modules consist of a means for performing the managing of the distributed execution of the programs for the clients.