OPTIMIZING DATA PARTITIONING FOR DATA-PARALLEL COMPUTING

• US 20130152057A1
• Filed: 12/13/2011
• Published: 06/13/2013
• Est. Priority Date: 12/13/2011
• Status: Active Grant

• Current Assignee: Microsoft Technology Licensing LLC (Microsoft Corporation)
• Original Assignee: Microsoft Corporation
• Inventors: Ke, Qifa, Prabhakaran, Vijayan, Xie, Yinglian, Yu, Yuan, Wu, Jingyue, Yang, Junfeng
• Granted Patent: US 9,235,396 B2
• Time in Patent Office: Days
• US Class Current: 717/132
• CPC Class Codes: G06F 8/453 Data distribution

Abstract

A data partitioning plan is automatically generated that—given a data-parallel program and a large input dataset, and without having to first run the program on the input dataset—substantially optimizes performance of the distributed execution system that explicitly measures and infers various properties of both data and computation to perform cost estimation and optimization. Estimation may comprise inferring the cost of a candidate data partitioning plan, and optimization may comprise generating an optimal partitioning plan based on the estimated costs of computation and input/output.

Claims

1. A system for optimizing data partitioning for a distributed execution engine, the system comprising:

a code/EPG analysis module for deriving properties of a data-parallel program code in each vertex in a corresponding execution plan graph (EPG) compiled from the data-parallel program code;

a complexity module for at least deriving the computational complexity of each vertex in the EPG;

a data analysis module for generating a plurality of compact data representations corresponding to an input data for processing by the data-parallel program code;

a statistics and samples module for determining the relationship between input data size versus computational and input-output (I/O) costs;

a cost modeling and estimation module for estimating the runtime cost of each vertex in the EPG and the overall runtime cost represented by the EPG; and

a cost optimization module for determining a data partitioning plan.

2. The system of claim 1, wherein the code/EPG analysis module and the data analysis module determine how to sample the input data.

3. The system of claim 1, wherein the code/EPG analysis module uses automatic static and dynamic programming language analysis techniques.

4. The system of claim 3, wherein the automatic static and dynamic programming language analysis techniques derive at least one semantic from among the following plurality of semantics:

processing flow, computational complexity, and relevant data features.

5. The system of claim 3, wherein the code/EPG analysis module uses hints for programming language analysis provided for user-defined functions.

6. The system of claim 1, further comprising an EPG module for compiling the data-parallel program code into the EPG, wherein the EPG comprises a plurality of vertices, and wherein each partition from among a plurality of initial data partitions is processed by at least one vertex from among the plurality of vertices.

7. The system of claim 6, wherein the EPG module produces the plurality of initial data partitions from the input data.

8. The system of claim 1, wherein the initial data partitions are produced by the EPG module using a partition function.

9. The system of claim 1, further comprising an updated EPG module for generating an updated EPG when the data partitioning plan is determined by the cost optimization module.

10. The system of claim 1, wherein the EPG is a directed acyclic graph having multiple stages wherein each stage comprises at least one vertex to process each input data partition from among a plurality of initial data partitions.

11. The system of claim 10, wherein at least two vertices run in parallel to process at least two initial data partitions.

12. A method for optimizing data partitioning for a distributed execution engine, the method comprising:

determining a plurality of parts of a data-parallel program code corresponding to each vertex in a corresponding execution plan graph (EPG), the EPG comprising a plurality of vertices corresponding to a plurality of initial data partitions;

deriving a computational complexity for each vertex from among the plurality of vertices in the EPG;

determining a plurality of relationships between a plurality of input data and a plurality of execution costs;

estimating a runtime cost for each vertex from among the plurality of vertices in the EPG; and

estimating the overall runtime cost represented by the EPG.

13. The method of claim 12, wherein determining the relationships between the input data and the execution costs is based on the determined parts of the data-parallel program code that correspond to each vertex in the corresponding EPG.

14. The method of claim 12, further comprising compiling a data-parallel program code into the EPG.

15. The method of claim 12, further comprising attempting to determine an improved data partitioning plan and:

if the improved data partitioning plan is determined, updating the EPG and iteratively repeating the estimating and determining until another improved data partitioning plan cannot be determined; and

if the improved data partitioning plan cannot be determined, outputting the resultant EPG.

16. The method of claim 15, further comprising producing the plurality of initial data partitions from the input data for processing by a data-parallel program code.

17. The method of claim 15, wherein determining the improved data partitioning plan comprises using a hierarchical partitioning graph.

18. A computer-readable medium comprising computer-readable instructions for optimizing data partitioning for a distributed execution engine, the computer-readable instructions comprising instructions that cause a processor to:

analyze a data-parallel program code and its corresponding execution plan graph (EPG);

analyze the input data and a plurality of corresponding initial data partitions in view of the results of analyzing the data-parallel program code and the EPG;

estimate a runtime cost for each vertex from among a plurality of vertices comprising the EPG;

determine an improved data partitioning plan and update the EPG accordingly; and

repeat the estimate and the determine until an optimized EPG is found.

19. The computer-readable medium of claim 18, further comprising instructions for causing the processor to:

deconstruct and recombine at least two data partitions into at least three or more smaller data partitions;

update the EPG;

identify a critical path in the updated EPG; and

split at least one vertex corresponding to the path in the updated EPG.

20. The computer-readable medium of claim 18, further comprising instructions for causing the processor to:

deconstruct and recombine at least two data partitions into at least three or more smaller data partitions;

update the EPG;

identify two or more smaller data partitions having smaller execution costs; and

merge the two or more smaller data partitions into a combined data partition.