High Level Programming Extensions For Distributed Data Parallel Processing

• US 20100241827A1
• Filed: 03/18/2009
• Published: 09/23/2010
• Est. Priority Date: 03/18/2009
• Status: Active Grant

• Current Assignee: Microsoft Technology Licensing LLC (Microsoft Corporation)
• Original Assignee: Microsoft Corporation
• Inventors: Erlingsson, Ulfar, McSherry, Frank, Yu, Yuan, Isard, Michael A.
• Granted Patent: US 8,209,664 B2
• Time in Patent Office: Days
• US Class Current: 712/30
• CPC Class Codes: G06F 8/314 Parallel programming languages G06F8/313 takes precedence
• 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 12/44 Star or tree networks

Abstract

General-purpose distributed data-parallel computing using high-level computing languages is described. 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. A set of extensions to a sequential high-level computing language are provided to support distributed parallel computations and to facilitate generation and optimization of distributed execution plans. The extensions are fully integrated with the programming language, thereby enabling developers to write sequential language programs using known constructs while providing the ability to invoke the extensions to enable better generation and optimization of the execution plan for a distributed computing environment.

Claims

1. A machine implemented method for distributed parallel processing, comprising:

accessing an expression from a sequential program that is executing at a first machine, the expression invoking at least one extension for distributed parallel processing by a distributed execution engine;

automatically generating an execution plan for parallel processing of the expression by the distributed execution engine using the at least one extension; and

providing the execution plan to the distributed execution engine for controlling parallel execution of the expression.

2. A machine implemented method according to claim 1, wherein:

the distributed execution engine includes a compute cluster having a plurality of nodes, the distributed execution engine controlling parallel execution of the expression in the compute cluster;

automatically generating the execution plan includes automatically generating a execution plan graph having vertices, automatically generating code based on the at least one extension and assigning the code to particular vertices in the execution plan graph, the automatically generating the execution plan being performed by a distributed execution provider; and

the method further comprises distributing the code to nodes in the compute cluster corresponding to the particular vertices in the execution plan graph.

3. A machine implemented method according to claim 2, wherein:

the extension is an operator specifying application of a user-defined function using data parallel processing, the user-defined function being in a high-level programming language; and

the expression includes a language integrated query.

4. A machine implemented method according to claim 3, wherein the user-defined function receives a first dataset and the automatically generating an execution plan includes:

specifying a partitioning of the first dataset at a set of nodes from said plurality of nodes;

determining whether the user-defined function includes an annotation that the function is operable on partitions of the first dataset;

if the user-defined function includes the annotation, automatically generating code for executing the user-defined function in parallel at the set of nodes; and

if the user-defined function does not include the annotation, automatically generating code for streaming partitions of the first dataset to a single node and code for executing the user-defined function at the single node.

5. A machine implemented method according to claim 4, further comprising:

if the user-defined function includes the annotation, outputting the code for executing the user-defined function in parallel to the set of nodes; and

if the user-defined function does not include the annotation, outputting the code for streaming partitions of the first dataset to the set of nodes and outputting the code for executing the user-defined function to the single node.

6. A machine implemented method according to claim 5, wherein the single node is part of the set of nodes.

7. A machine implemented method according to claim 3, wherein the automatically generating an execution plan includes:

determining whether the user-defined function includes an annotation specifying the function is not stateful;

if the user-defined function includes the annotation specifying the function is not stateful, automatically generating a first version of the execution plan; and

if the user-defined function does not include the annotation specifying the function as not stateful, automatically generating a second version of the execution plan.

8. A machine implemented method according to claim 1, wherein the expression references a first dataset, the method further comprising:

determining whether the first dataset includes a partitioning property;

if the first dataset includes a partitioning property, optimizing the execution plan such that the first dataset is not re-partitioned in the compute cluster; and

if the first dataset does not include a partitioning property, specifying a partitioning of the first dataset in the execution plan and dividing the first dataset into individual partitions at nodes of the compute cluster according to the specified partitioning.

9. A machine implemented method according to claim 1, wherein:

the expression references a first dataset, the first dataset of a type having a first field;

automatically generating the execution plan includes determining whether the first field includes an annotation that the first field is not nullable; and

if the first field includes the annotation that the first field is not nullable, optimizing the execution plan based on presence of the first field not being nullable.

10. A machine implemented method according to claim 1, wherein:

the expression references a first dataset, the first dataset of a type having a first field;

automatically generating the execution plan includes determining whether the expression includes an annotation mapping the first field to a second field of a second dataset; and

if the first field is mapped to a second field of a second dataset, generating the execution plan to access an output of a second expression referencing the second dataset to determine a result of evaluating the first expression.

11. A distributed parallel processing system, comprising:

a compute cluster including a plurality of nodes, each node including at least one processor;

an execution provider that accesses expressions from a sequential program that is running at a first machine, the execution provider receives at least one expression specifying an extension for distributed data parallel processing in the compute cluster, the execution provider automatically generates an execution plan graph and code for vertices in the execution plan graph using the extension for distributed data parallel processing; and

an execution engine that receives the execution plan graph and the code from the execution provider and manages parallel execution of the expression in the compute cluster based on the execution plan graph and the code.

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

the extension is an operator specifying division of a first dataset into multiple datastreams at the compute cluster, the code for at least one of the vertices receives the first dataset and provides a second dataset having at least two datastreams included therein, the two datastreams including records from the first dataset; and

the execution engine distributes the code for the at least one of the vertices receiving the first dataset to one or more nodes in the compute cluster corresponding to the at least one of the vertices.

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

the execution provider accesses a first expression and a second expression from the program running at the first machine, the first expression generating a first output and the second expression generating a second output, the first expression invoking a first operating specifying deferred execution of the first expression in the compute cluster and the second expression invoking the first operator specifying deferred execution of the second expression in the compute cluster;

the execution provider accesses a third expression from the program invoking a second operator calling for generation of the first output and the second output;

the execution provider generates the execution plan graph and the code for vertices of the execution plan graph for a single job including the first expression, the second expression and the third expression; and

the execution engine generates the first output and the second output from the compute cluster as the single job.

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

the execution provider accesses a first expression from the program running at the first machine, the first expression invoking a method specifying that the first dataset has been sorted; and

the execution provider optimizes the execution plan graph based on the specified sorting of the first dataset.

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

the execution provider accesses a first expression from the program running at the first machine, the first expression invoking a method specifying that each record of the first dataset is distinct; and

the execution provider optimizes the execution plan graph based on each record of the first dataset being distinct.

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

receiving one or more expressions from an application executing at a first machine, the one or more expressions referencing a first dataset;

automatically generating an execution plan for executing the one or more expressions in parallel at nodes of a compute cluster, the automatically generating including;

determining whether the one or more expressions include an extension specifying a particular data partitioning for the first dataset,

if the one or more expressions include the extension specifying a particular data partitioning, generating a first execution graph with the first dataset partitioned according to the particular data partitioning, and

if the one or more expressions do not include the extension specifying a particular data partitioning, generating a second execution graph with the first dataset partitioned according to a different data partitioning; and

providing the execution plan to an execution engine that controls parallel execution of the one or more expressions in the compute cluster.

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

determining whether the one or more expressions include an extension specifying a particular data partitioning includes determining whether the one or more expressions include an extension specifying either hash-partitioning or range-partitioning.

18. One or more processor readable storage devices according to claim 16, wherein the method further comprises:

dividing the first dataset into a plurality of partitions according to the particular data partitioning or the different data partitioning;

providing each partition of the plurality to one node in the compute cluster; and

executing the one or more expressions in parallel at the nodes of the compute cluster.

19. One or more processor readable storage devices according to claim 16, wherein automatically generating an execution plan further includes:

determining whether the one or more expressions include a property indicating that the first dataset is already partitioned; and

if the one or more expressions include the property, generating the execution plan to distribute the first dataset in the compute cluster according to the partitioning indicated by the property.

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

p1 the one or more expressions include an aggregate function invoking a user-defined function, the user-defined function referencing the first dataset; and

the automatically generating an execution plan includes;

determining whether the user-defined function includes an annotation specifying the user-defined function as associative,

if the user-defined function includes the annotation, generating code to apply the aggregate function on the individual partitions of the first dataset and code to combine the results of applying the aggregate function on the individual partitions, and

if the user-defined function does not include the annotation, generating code to stream the individual partitions of the first dataset to a single node for application of the aggregate function.