Parallel computing execution plan optimization

Patent metadata

Field	Value
Publication number	US-9235446-B2
Application number	US-201213530490-A
Country	US
Kind code	B2
Filing date	Jun 22, 2012
Priority date	Jun 22, 2012
Publication date	Jan 12, 2016
Grant date	Jan 12, 2016

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.

First claim

What is claimed is:

1. A computer storage 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, wherein the optimizing includes at least one of:
reducing a number of parallelisms for operations at a tail end of the job; or
replacing pair-wise joins of data with broadcast joins of data process by at least one of the operations.

2. The computer storage 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 storage 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 storage 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 storage 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 storage 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 storage 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 storage 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.

9. The computer storage medium of claim 8 , 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.

10. The computer storage 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.

11. 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, wherein the optimizing includes at least one of:
reducing a number of parallelisms for operations at a tail end of the job; or
replacing pair-wise joins of data with broadcast joins of data process by at least one of the operations;

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.

12. The computer-implemented method of claim 11 , 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.

13. The computer-implemented method of claim 11 , 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.

14. The computer-implemented method of claim 11 , 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.

15. The computer-implemented method of claim 11 , wherein the collecting the statistics includes:
selecting a set of operations from the operations performed in the tasks by ignoring at least one operation that produces a predictable result;
inserting collectors into the set of operations; and
receiving at least some of the statistics from the collectors.

16. 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
a 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, wherein the compile-time optimizer is configured to perform at least one of:
reducing a number of parallelisms for operations at a tail end of the job; or
replacing pair-wise joins of data with broadcast joins of data process by at least one of the operations.

17. The computing device of claim 16 , wherein the compiler generates the second e

Assignees

• Bruno Nicolas
• Zhou Jingren
• Kandula Srikanth
• Agarwal Sameer
• Wu Ming-Chuan
• Microsoft Technology Licensing Llc

Inventors

• Bruno Nicolas
• Zhou Jingren
• Kandula Srikanth
• Agarwal Sameer
• Wu Ming-Chuan

Classifications

• G06F9/5066 Primary

  Algorithms for mapping a plurality of inter-dependent sub-tasks onto a plurality of physical CPUs (mappping at compile time, see G06F8/451) · CPC title

• G06F16/24

  Querying · CPC title