General purpose distributed data parallel computing using a high level language

• US 9,110,706 B2
• Filed: 02/09/2009
• Issued: 08/18/2015
• Est. Priority Date: 02/09/2009
• Status: Active Grant

• Current Assignee: Microsoft Technology Licensing LLC (Microsoft Corporation)
• Original Assignee: Microsoft Technology Licensing LLC (Microsoft Corporation)
• Inventors: Yu, Yuan, Fetterly, Dennis, Isard, Michael, Erlingsson, Ulfar, Budiu, Mihai
• Primary Examiner(s): Breene, John
• Assistant Examiner(s): Dinh, Lynda
• Application Number:
  US12/368,231
• Publication Number:
  US 20100205588A1
• Time in Patent Office: 2,381 Days
• Field of Search: 717/149, 717/153, 709/203
• US Class Current: 1/1
• CPC Class Codes: G06F 3/00 Input arrangements for transferring data to be processed into a form capable of being handled by the computer; Output arrangements for transferring data from processing unit to output unit, e.g. interface arrangements
• CPC Class Codes: G06F 7/00 Methods or arrangements for processing data by operating upon the order or content of the data handled logic circuits H03K19/00
• CPC Class Codes: G06F 8/20 Software design
• CPC Class Codes: G06F 8/24 Object-oriented
• CPC Class Codes: G06F 8/433 Dependency analysis; Data or control flow analysis
• CPC Class Codes: G06F 8/4434 Reducing the memory space required by the program code
• CPC Class Codes: G06F 8/4441 Reducing the execution time required by the program code
• CPC Class Codes: G06F 8/445 Exploiting fine grain parallelism, i.e. parallelism at instruction level run-time instruction scheduling G06F9/3836
• CPC Class Codes: G06F 8/456 Parallelism detection
• CPC Class Codes: G06F 8/458 Synchronisation, e.g. post-wait, barriers, locks synchronisation among tasks G06F9/52
• CPC Class Codes: G06F 9/52 Program synchronisation; Mutual exclusion, e.g. by means of semaphores

Abstract

General-purpose distributed data-parallel computing using a high-level language is disclosed. Data parallel portions of a sequential program that is written by a developer in a high-level language are automatically translated into a distributed execution plan. The distributed execution plan is then executed on large compute clusters. Thus, the developer is allowed to write the program using familiar programming constructs in the high level language. Moreover, developers without experience with distributed compute systems are able to take advantage of such systems.

Claims

1. A system, comprising:

a processor readable storage hardware device;

a user interface; and

a processor coupled to the processor readable hardware storage device and to the user interface, wherein the processor readable hardware storage device includes instructions that cause the processor to;

execute a sequential application program comprising a data parallel portion that includes an expression, wherein the application is written in a high-level language and comprises both imperative operations and declarative operations;

access the expression from a portion of the sequential application program that comprises a declarative operation;

based on the expression, automatically generate an execution plan graph, the execution plan graph including a directed graph having vertices that represent processes and edges between the vertices that represent data channels, the execution plan graph for executing the expression in parallel at nodes of a compute cluster, including causing the processor to break the expression into a plurality of sub-expressions, each of the sub-expressions is a vertex in the directed graph;

automatically generate vertex code for the vertices of the execution plan graph;

automatically generate serialization code that allows data to be passed in the data channels between the vertices;

provide the execution plan graph, the serialization code, and the vertex code to an execution engine of the compute cluster that manages parallel execution of the expression in the compute cluster based on the execution plan graph, the serialization code, and the vertex code;

receive results of executing the execution plan graph in the compute cluster; and

execute a portion of the sequential application program that comprises an imperative operation to present the results in the user interface.

2. The system of claim 1, wherein the processor further serializes data objects that are referenced in the sequential application program and that are needed for execution of the vertex code and provides the serialized data objects to the execution engine of the compute cluster.

3. The system of claim 1, wherein:

the processor further generates callback code for optimizing execution in the compute cluster and provides the callback code to the execution engine of the compute cluster.

4. The system of claim 1, wherein the instructions further cause the processor to provide code to the compute cluster for performing dynamic optimizations based on information collected while the expression is executing in the compute cluster.

5. The system of claim 1, wherein the instructions further cause the processor to:

automatically generate code for performing dynamic optimizations, wherein the code for performing dynamic optimizations comprises API hooks to dynamically mutate the execution plan as information from running the execution plan in the compute cluster becomes available; and

provide the code for performing dynamic optimizations to the execution engine in the compute cluster.

6. The system of claim 1, wherein the instructions further cause the processor to create local table objects that encapsulate output of execution of the expression in the compute cluster, the local table objects are accessible by the sequential application.

7. A machine implemented method comprising:

executing, by a processor on a client machine, a sequential application program comprising a data parallel portion that includes an expression, wherein the application is written in a high-level language and comprises both imperative operations and declarative operations;

accessing, by the processor, an expression from a portion of the sequential application program that comprises a declarative operation;

based on the expression, the processor automatically generating an execution plan graph, the execution plan graph including a directed graph having vertices that represent processes and edges between the vertices that represent data channels, the execution plan graph for executing the expression in parallel at nodes of a compute cluster, including the processor breaking the expression into a plurality of sub-expressions, wherein each of the sub-expressions is a vertex in the directed graph;

automatically generating vertex code for the vertices of the execution plan graph by the processor;

automatically generating serialization code that allows data to be passed in the data channels between the vertices by the processor;

providing, from the client machine to an execution engine of the compute cluster, the execution plan graph, the serialization code, and the vertex code, wherein the execution engine manages parallel execution of the expression in the compute cluster based on the execution plan graph, the serialization code, and the vertex code;

receiving, by the client machine, results of executing the execution plan graph in the compute cluster; and

executing, by the processor, a portion of the sequential application program that comprises an imperative operation to present the results in a user interface associated with the client machine.

8. The method of claim 7, further comprising the processor serializing data objects that are referenced in the sequential application program and that are needed for execution of the vertex code and providing the serialized data objects to the execution engine of the compute cluster.

9. The method of claim 7, further comprising the processor generating callback code for optimizing execution in the compute cluster and providing the callback code to the execution engine of the compute cluster.

10. The method of claim 7, wherein the expression includes tree nodes that are operators and child tree nodes that are inputs to the operators.

11. A method of claim 7, further including the processor optimizing the execution plan at the client machine based on inferred properties.

12. A machine implemented method as recited in claim 11, wherein the optimizing the execution plan is a static optimization performed at the client machine.

13. A machine implemented method as recited in claim 7, further including the processor providing code for performing dynamic optimizations to the execution plan based on information collected while the expression is executing in the compute cluster, wherein the code for performing dynamic optimizations is provided from the client machine to the compute cluster.

14. A machine implemented method as recited in claim 7, further including:

automatically generating, by the processor, code for performing dynamic optimizations, wherein the code for performing dynamic optimizations comprises API hooks to dynamically mutate the execution plan as information from running the execution plan in the compute cluster becomes available; and

providing the code for performing dynamic optimizations from the client machine to the execution engine in the compute cluster.

15. The machine implemented method as recited in claim 7, further comprising creating local table objects on the client machine that encapsulate output of execution of the expression in the compute cluster, wherein the local table objects are accessible by the sequential application.

16. A machine implemented method as recited in claim 7, further including:

receiving, by the processor, information pertaining to a failure during execution of the expression in the compute cluster; and

executing, by the processor, the expression in isolation at the client machine to analyze the failure.

17. A computer readable storage hardware device having stored thereon computer executable instructions which, when executed on a processor, cause the processor to:

execute a sequential application program comprising a data parallel portion that includes an expression, wherein the application is written in a high-level language and comprises both imperative operations and declarative operations;

access the expression from a portion of the sequential application program that comprises a declarative operation;

based on the expression, automatically generate an execution plan graph, the execution plan graph including a directed graph having vertices that represent processes and edges between the vertices that represent data channels, the execution plan graph for executing the expression in parallel at nodes of a compute cluster, including causing the processor to break the expression into a plurality of sub-expressions, each of the sub-expressions is a vertex in the directed graph;

automatically generate vertex code for the vertices of the execution plan graph;

automatically generate serialization code that allows data to be passed in the data channels between the vertices;

provide the execution plan graph, the serialization code, and the vertex code to an execution engine of the compute cluster that manages parallel execution of the expression in the compute cluster based on the execution plan graph, the serialization code, and the vertex code;

receive results of executing the execution plan graph in the compute cluster; and

execute a portion of the sequential application program that comprises an imperative operation to present the results in a user interface.

18. The computer readable storage hardware device of claim 17, wherein the computer executable instructions further cause the processor to generate callback code for optimizing execution in the compute cluster and provides the callback code to the execution engine of the compute cluster.

19. The computer readable storage hardware device of claim 17, wherein the computer executable instructions further cause the processor to:

automatically determine data objects that are referenced in the sequential application program and are needed for execution of the vertex code for each of the sub-expressions, the data objects having certain data types;

serialize the referenced data objects; and

output the serialized data objects to the execution engine of the compute cluster.

20. The computer readable storage hardware device of claim 17, wherein the computer executable instructions further cause the processor to create local table objects that encapsulate output of execution of the expression in the compute cluster, wherein the local table objects are accessible by the sequential application.