DISTRIBUTED PARALLEL COMPUTING

• US 20080082644A1
• Filed: 09/29/2006
• Published: 04/03/2008
• Est. Priority Date: 09/29/2006
• Status: Abandoned Application

• Current Assignee: Microsoft Technology Licensing LLC (Microsoft Corporation)
• Original Assignee: Microsoft Corporation
• Inventors: Fetterly, Dennis C., Yu, Yuan, Isard, Michael A., Birrell, Andrew D., Budiu, Mihai-Dan
• Application Number:
  US11/537,506
• Publication Number:
  US 20080082644A1
• Time in Patent Office: Days
• US Class Current: 709/223
• 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: H04L 67/1097 for distributed storage of data in networks, e.g. transport arrangements for network file system [NFS], storage area networks [SAN] or network attached storage [NAS]

Abstract

A general purpose high-performance distributed execution engine for coarse-grained data-parallel applications is proposed that allows developers to easily create large-scale distributed applications without requiring them to master concurrency techniques beyond being able to draw a graph of the data-dependencies of their algorithms. Based on the graph, a job manager intelligently distributes the work load so that system resources are used efficiently. The system is designed to scale from a small cluster of a few computers, or the multiple CPU cores on a powerful single computer, up to a data center containing thousands of servers.

Claims

1. A method for performing distributed parallel processing, comprising:

finding one or more locations for each of a set of data files in a distributed data processing system having multiple nodes, said data files are for use by a set of program units;

identifying nodes of said distributed data processing system that are available for executing said program units and are near said data files; and

sending instructions to nodes near said data files to execute said program units.

2. A method according to claim 1, wherein:

said set of nodes includes a first node and a set of other nodes, said first node is a single computing machine and said set of other nodes are multiple computing machines; and

said sending instructions includes sending instructions to said first node to concurrently execute a first subset of said program units on said first node and sending instructions to said other nodes to concurrently execute a second subset of said program units across said other nodes.

3. A method according to claim 1, wherein:

said multiple nodes include multiple computing machines connected to a network with sets of computing machines separated by network switches; and

a particular node is near one of said data files if said particular node is connected to a common switch with a storage device that is storing said data file.

4. A method according to claim 1, wherein:

said program units implement vertices on a user customizable directed acyclic graph that are part of a system that counts frequency of items in network search data stored across said data files.

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

evaluating code that provides a definition of a user customizable directed acyclic graph; and

building said graph based on said code, said code includes said program units, said program units implement a vertices in said graph, said step of sending instructions is performed based on said graph.

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

providing fault tolerance for said program units, said data files are stored on said nodes.

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

providing said program units to said nodes near said data files if said nodes near said data files do not have a local copy of said program units.

8. A method according to claim 1, wherein:

said multiple nodes include multiple computing machines;

said set of data files include redundant copies of a set of data;

each data file stores a portion of one copy of said set of data;

said data files are distributed across two or more of said machines; and

said sending instructions comprises sending said instructions to nodes that are storing said data files.

9. A distributed parallel processing system, comprising:

a job manager, said job manager manages execution of a system defined by a user customizable directed acyclic graph, said user customizable graph includes a set of vertices corresponding to a set of program units; and

a plurality of computing machines in communication with said job manager, said plurality of computing machines include a first computing machine and a set of other machines, said first machine concurrently runs a first subset of multiple program units of said set of program units that correspond to multiple vertices of said graph, said set of other machines concurrently runs a second subset of multiple program units of said set of program units across different machines, said second subset of multiple program units correspond to multiple vertices of said graph.

10. A distributed parallel processing system according to claim 9, wherein:

said program units access data distributed across at least two or more of said computing machines.

11. A distributed parallel processing system according to claim 9, wherein:

said set of program units determine a statistic with respect to network search data;

said network search data is divided into multiple units, each unit is stored in redundant files; and

said job manager chooses which of said computing machines to execute specific program units based on proximity of said computing machines to said redundant files.

12. A distributed parallel processing system according to claim 9, wherein:

said first machine runs said first subset of multiple program units using different threads on a single computing machine.

13. A distributed parallel processing system according to claim 9, wherein:

said job manager treats said first subset of multiple program units as a single vertex for purposes of data flow external to said first machine.

14. A distributed parallel processing system according to claim 9, further comprising:

a name server, said name server provides identification and location information to said job manager for said plurality of computing machines.

15. A distributed parallel processing system according to claim 9, wherein:

said first subset of multiple program units communicate with each other using a shared memory FIFO.

16. A distributed parallel processing system according to claim 9, wherein:

said second subset of multiple program units communicate with each other using a TCP/IP pipe.

17. One or more processor readable storage devices having processor readable code stored thereon, said processor readable code programs one or more processors to perform a method comprising:

evaluating a definition of a customizable directed acyclic graph;

building said graph based on said definition;

accessing first code that implements a vertex on said graph, said code implements a portion of a system that counts frequency of items in a network search data unit;

identifying multiple nodes that store said network search data unit;

determining that a particular node of said multiple nodes is best suited for executing said first code; and

sending instructions to execute said first code to said particular node because said particular node was determined to be best suited for executing said code.

18. One or more processor readable storage devices according to claim 17, wherein:

said particular node was determined to be best suited for executing said first code because said particular node and a data store that stores data for said first code are located in a common region of a network associated with a particular switch device.

19. One or more processor readable storage devices according to claim 17, wherein:

said method further comprises providing results from said first code to second code on a different node of said multiple nodes; and

said multiple nodes are separate computing devices on a network.

20. One or more processor readable storage devices according to claim 17, wherein:

said first code is object oriented code.