PARALLEL DATA COMPUTING OPTIMIZATION

• US 20130346988A1
• Filed: 06/22/2012
• Published: 12/26/2013
• Est. Priority Date: 06/22/2012
• Status: Active Grant

• Current Assignee: Microsoft Technology Licensing LLC (Microsoft Corporation)
• Original Assignee: Microsoft Corporation
• Inventors: Bruno, Nicolas, Zhou, Jingren, Kandula, Srikanth, Wu, Ming-Chuan, Agarwal, Sameer
• Granted Patent: US 9,235,446 B2
• Time in Patent Office: Days
• US Class Current: 718/102
• CPC Class Codes: G06F 16/24 Querying
• 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

Abstract

The use of statistics collected during the parallel distributed execution of the tasks of a job may be used to optimize the performance of the task or similar recurring tasks. An execution plan for a job is initially generated, in which the execution plan includes tasks. Statistics regarding operations performed in the tasks are collected while the tasks are executed via parallel distributed execution. Another execution plan is then generated for another recurring job, in which the additional execution plan has at least one task in common with the execution plan for the job. The additional execution plan is subsequently optimized based at least on the statistics to produce an optimized execution plan.

Claims

1. A computer-readable medium storing computer-executable instructions that, when executed, cause one or more processors to perform acts comprising:

generating a first execution plan for a job that includes a plurality of tasks;

collecting statistics regarding operations performed in the plurality of tasks while the plurality of tasks are executed via parallel distributed execution;

generating a second execution plan for a recurring job, the second execution plan having at least one task in common with the first execution plan for the job; and

optimizing the second execution plan based at least on the statistics to produce an optimized execution plan.

2. The computer-readable medium of claim 1, further comprising:

executing a plurality of further tasks specified in the optimized execution plan to generate execution results for the recurring job; and

outputting the execution results of the recurring job.

3. The computer-readable medium of claim 2, further comprising:

collecting additional statistics regarding additional operations performed in the further tasks while the further tasks are executed via parallel distributed execution; and

optimizing the additional operations on-the-fly during the parallel distributed execution of the further tasks based at least on the additional statistics.

4. The computer-readable medium of claim 1, wherein the generating the first execution plan includes translating a job script for the job into the first execution plan.

5. The computer-readable medium of claim 1, wherein the collecting the statistics includes:

inserting collectors into a set of operations selected from the operations performed in the tasks;

receiving local statistical data from the collectors while the set of the operations are executed via parallel distributed execution; and

composing the local statistical data provided by the collectors into the statistics.

6. The computer-readable medium of claim 5, further comprising selecting the set of operations from the operations performed in the task by ignoring at least one operation that produces a predictable result.

7. The computer-readable medium of claim 5, wherein the optimizing includes matching the statistics captured by the collectors to expressions specified in the second execution plan based at least on hash values that represent locations of the collectors in a query graph of the first execution plan.

8. The computer-readable medium of claim 1, wherein the optimizing includes at least one of:

reducing a number of parallelisms for operations at a tail end of the job;

performing re-partitioning of data on an as-needed basis;

combining at least two operations at the tail end of the job into a single operation;

or

replacing pair-wise joins of data with broadcast joins of data process by at least one of the operations.

9. The computer-readable medium of claim 1, wherein the statistics include a first number of distinctive values in a data table column and a second number of values that repeat over a predetermined fraction of rows in the data table column, and wherein the optimizing includes avoiding skewing in partitioning of operations in the recurring job based at least on the first number and the second number.

10. The computer-readable medium of claim 9, further comprising:

estimating the first number via distributed hash sketching; and

estimating the second number by at least receiving frequency estimates obtained by each collector, totaling up the frequency estimates, and tallying up distinct values, wherein each frequency estimate is obtained by a corresponding collector through lossy counting of a subset of data observed by a task that is monitored by the corresponding collector.

11. The computer-readable medium of claim 1, wherein the optimizing includes at least one of:

determining a sequence of operations based at least on operator selectivity and computation costs of the operations;

or

determining a degree of parallelism for the operations, a number of partitions for the operations, or operations to fit into a particular task based at least on the computation costs of the operations and a size of data to be processed.

12. A computer-implemented method, comprising:

generating a first execution plan for a job that includes a plurality of tasks;

collecting statistics regarding operations performed in the plurality of tasks while the plurality of tasks are executed via parallel distributed execution;

optimizing a second execution plan for a recurring job based at least on the statistics to produce an optimized execution plan, the second execution plan having at least one task in common with the first execution plan;

executing a plurality of further tasks specified in the optimized execution plan to generate execution results for the recurring job; and

outputting the execution results of the recurring job.

13. The computer-implemented method of claim 12, further comprising:

collecting additional statistics regarding additional operations performed in the further tasks while the further tasks are executed via parallel distributed execution; and

optimizing the additional operations on-the-fly during the parallel distributed execution of the further tasks based at least on the additional statistics.

14. The computer-implemented method of claim 12, wherein the collecting the statistics includes:

inserting collectors into a set of operations selected from the operations performed in the tasks;

receiving local statistical data from the collectors while the set of the operations are executed via parallel distributed execution; and

composing the local statistical data provided by the collectors into the statistics.

15. The computer-implemented method of claim 12, wherein the optimizing includes at least one of:

reducing a number of parallelisms for operations at a tail end of the job;

combining at least two operations at the tail end of the job into a single operation;

or

replacing pair-wise joins of data with broadcast joins of data process by at least one of the operations.

16. The computer-implemented method of claim 12, wherein the optimizing includes at least one of:

determining a sequence of operations based at least on operator selectivity and computation costs of the operations;

or

determining a degree of parallelism for the operations, a number of partitions for the operations, or operations to fit into a particular task based at least on the computation costs of operations.

17. A computing device, comprising:

one or more processors; and

a memory that includes a plurality of computer-executable components executed by the one or more processors, comprising;

a compiler that generates a first execution plan for a job that includes a plurality of tasks;

at least one job manager that collects statistics regarding operations performed in the plurality of tasks while the plurality of tasks are executed via parallel distributed execution; and

an compile-time optimizer that optimizes a second execution plan based at least on the statistics to produce an optimized execution plan, the second execution plan having at least one task in common with the first execution plan.

18. The computing device of claim 17, wherein the compiler generates the second execution plan for a recurring job that is similar to the job.

19. The computing device of claim 17, wherein the at least one job manager collects additional statistics regarding additional operations performed in further tasks of the second execution plan as the further tasks are executed via parallel distributed execution, and optimizes the additional operations on-the-fly during the parallel distributed execution of the further tasks based at least on the additional statistics.

20. The computing device of claim 17, wherein the at least one job manager collects the statistics by:

insert collectors into a set of operations selected from the operations performed in the tasks;

receive local statistical data from the collectors as the set of the operations are executed via parallel distributed execution; and

compose the local statistical data provided by the collectors into the statistics.