System and method for limiting the impact of stragglers in large-scale parallel data processing

• US 8,510,538 B1
• Filed: 04/13/2010
• Issued: 08/13/2013
• Est. Priority Date: 04/13/2009
• Status: Active Grant

• Current Assignee: Google LLC (Alphabet Inc.)
• Original Assignee: Google Inc. (Alphabet Inc.)
• Inventors: Malewicz, Grzegorz, Dvorsky, Marian, Colohan, Christopher B., Thomson, Derek P., Levenberg, Joshua Louis
• Primary Examiner(s): Faherty, Corey S
• Application Number:
  US12/759,637
• Time in Patent Office: 1,218 Days
• Field of Search: 712/220
• US Class Current: 712/225
• CPC Class Codes: G06F 15/00 Digital computers in general details G06F1/00 – G06F13/00; Data processing equipment in general
• CPC Class Codes: G06F 16/278 Data partitioning, e.g. horizontal or vertical partitioning
• CPC Class Codes: G06F 7/38 Methods or arrangements for performing computations using exclusively denominational number representation, e.g. using binary, ternary, decimal representation
• CPC Class Codes: G06F 9/00 Arrangements for program control, e.g. control units program control for peripheral devices G06F13/10
• CPC Class Codes: G06F 9/44 Arrangements for executing specific programs
• CPC Class Codes: G06F 9/5077 Logical partitioning of resources; Management or configuration of virtualized resources specific details on emulation or internal functioning of virtual machines G06F9/455
• CPC Class Codes: G06F 9/54 Interprogram communication

Abstract

A large-scale data processing system and method including a plurality of processes, wherein a master process assigns input data blocks to respective map processes and partitions of intermediate data are assigned to respective reduce processes. In each of the plurality of map processes an application-independent map program retrieves a sequence of input data blocks assigned thereto by the master process and applies an application-specific map function to each input data block in the sequence to produce the intermediate data and stores the intermediate data in high speed memory of the interconnected processors. Each of the plurality of reduce processes receives a respective partition of the intermediate data from the high speed memory of the interconnected processors while the map processes continue to process input data blocks an application-specific reduce function is applied to the respective partition of the intermediate data to produce output values.

Claims

1. A method of performing a large-scale data processing job, comprising:

executing a plurality of processes on a plurality of interconnected processors, the plurality of processes including a master process for coordinating a data processing job for processing a set of input data, and plurality of map processes and a plurality of reduce processes;

in the master process, assigning input data blocks of a set of input data to respective map processes of the plurality of map processes and assigning partitions of intermediate data to respective reduce processes of the plurality of reduce processes;

in each of the plurality of map processes;

executing an application-independent map program to retrieve a sequence of input data blocks assigned thereto by the master process and to apply an application-specific map function to each input data block in the sequence to produce the intermediate data; and

storing the intermediate data in memory of the interconnected processors; and

in each of the plurality of reduce processes;

receiving a respective partition of the intermediate data from the memory of the interconnected processors; and

applying an application-specific reduce function to the respective partition of the intermediate data to produce output values; and

in a respective reduce process;

receiving multiple distinct partitions of the intermediate data and

processing the multiple partitions one at a time in succession; and

identifying the respective reduce process as a reduce process that is delaying the data processing job while continuing to process intermediate data and, in response, reassigning at least one of the multiple partitions, which has not yet been processed, to a second reduce process, including copying the intermediate data in the reassigned partition to the other reduce process.

2. The method of claim 1, further comprising sorting the intermediate data into the plurality of partitions of the intermediate data.

3. The method of claim 2, wherein the data processing job is initiated by a user, and the intermediate data is sorted into the plurality of partitions based on an application-specific partition function selected by the user.

4. The method of claim 3, wherein the application-specific partition function is defined by the user.

5. The method of claim 1, wherein the data processing job is initiated by a user, and the application-specific map function and the application-specific reduce function are selected by the user.

6. The method of claim 5, wherein the application-specific map function and the application-specific reduce function are defined by the user.

7. The method of claim 1, wherein:

producing the intermediate data includes producing a plurality of blocks of intermediate data, wherein each block of intermediate data includes all of the intermediate data produced by applying the application-specific map function to a respective block of input data; and

receiving a respective partition of the intermediate data includes receiving a subset of the intermediate data in a first block of intermediate data that is associated with the respective partition while a second block of intermediate data is being produced, the second block of intermediate data including at least some intermediate data that is associated with the respective partition.

8. The method of claim 1, further comprising identifying a partition that is likely to delay the data processing job using predefined criteria and taking a remedial action.

9. The method of claim 8, wherein identifying a partition that is likely to delay the data processing job includes determining the size of the partition relative to the size of other partitions in the data processing job.

10. The method of claim 8, wherein remedial action comprises scheduling the partition for processing on a high capacity reduce process.

11. The method of claim 1, wherein the intermediate data in the reassigned partition is copied from memory associated with the respective reduce process.

12. The method of claim 1, further comprising, after identifying the respective reduce process as a reduce process that is delaying the data processing job, dividing the intermediate data in a partition that is assigned to the respective reduce process into a plurality of subpartitions and assigning each subpartition to a reduce process that is not the respective reduce process.

13. The method of claim 12, wherein dividing the intermediate data in the partition that is assigned to the respective reduce process includes copying the intermediate data in the partition from memory associated with the respective reduce process to memory associated with a reduce process that is not the respective reduce process.

14. The method of claim 1, wherein applying an application-specific reduce function to the respective partition of the intermediate data to produce output values includes:

while continuing to receive a respective partition of the intermediate data;

storing at least a subset of the intermediate data of the respective partition in memory associated with the reduce process;

while the intermediate data is stored in the memory associated with the reduce process, applying an application-specific combiner function to produce combined intermediate data values; and

applying the application-specific reduce function to the combined intermediate data values to produce output values.

15. The method of claim 14, wherein the combiner function is the same function as the application-specific reduce function.

24. The method of claim 1, wherein receiving the respective partition of the intermediate data from the memory of the interconnected processors occurs while the map processes that produced the received intermediate data continue to process input data blocks.

16. A system for large-scale processing of data, comprising:

memory;

one or more processors; and

one or more modules stored in the memory and executed by the one or more processors, the one or more modules including instructions to;

execute a plurality of processes on a plurality of interconnected processors, the plurality of processes including a master process for coordinating a data processing job for processing a set of input data, and plurality of map processes and a plurality of reduce processes;

in the master process, assign input data blocks of a set of input data to respective map processes of the plurality of map processes and assigning partitions of intermediate data to respective reduce processes of the plurality of reduce processes;

in each of the plurality of map processes;

execute an application-independent map program to retrieve a sequence of input data blocks assigned thereto by the master process and to apply an application-specific map function to each input data block in the sequence to produce the intermediate data; and

store the intermediate data in memory of the interconnected processors; and

in each of the plurality of reduce processes;

receive a respective partition of the intermediate data from the memory of the interconnected processors; and

apply an application-specific reduce function to the respective partition of the intermediate data to produce output values; and

in a respective reduce process;

receive multiple distinct partitions of the intermediate data and

process the multiple partitions one at a time in succession; and

identify the respective reduce process as a reduce process that is delaying the data processing job while continuing to process intermediate data and, in response, reassign at least one of the multiple partitions, which has not yet been processed, to a second reduce process, including copying the intermediate data in the reassigned partition to the other reduce process.

17. The system of claim 16, wherein:

the instructions to produce the intermediate data include instructions to produce a plurality of blocks of intermediate data, wherein each block of intermediate data includes all of the intermediate data produced by applying the application-specific map function to a respective block of input data; and

the instructions to receive a respective partition of the intermediate data include instructions to receive a subset of the intermediate data in a first block of intermediate data that is associated with the respective partition while a second block of intermediate data is being produced, the second block of intermediate data including at least some intermediate data that is associated with the respective partition.

18. The system of claim 16, further comprising instructions,

responsive to identifying the respective reduce process as a reduce process that is delaying the data processing job, to divide the intermediate data in a partition that is assigned to the respective reduce process into a plurality of subpartitions and assign each subpartition to a reduce process that is not the respective reduce process.

19. The system of claim 16, wherein the instructions to apply an application-specific reduce function to the respective partition of the intermediate data to produce output values include instructions to:

while continuing to receive a respective partition of the intermediate data;

store at least a subset of the intermediate data of the respective partition in memory associated with the reduce process;

while the intermediate data is stored in the memory associated with the reduce process, apply an application-specific combiner function to produce combined intermediate data values; and

apply the application-specific reduce function to the combined intermediate data values to produce output values.

25. The system of claim 16, wherein receiving the respective partition of the intermediate data from the memory of the interconnected processors occurs while the map processes that produced the received intermediate data continue to process input data blocks.

20. A non-transitory computer readable storage medium storing one or more programs for execution by one or more processors of a client device, the one or more programs comprising instructions to:

execute a plurality of processes on a plurality of interconnected processors, the plurality of processes including a master process for coordinating a data processing job for processing a set of input data, and plurality of map processes and a plurality of reduce processes;

in the master process, assign input data blocks of a set of input data to respective map processes of the plurality of map processes and assigning partitions of intermediate data to respective reduce processes of the plurality of reduce processes;

in each of the plurality of map processes;

execute an application-independent map program to retrieve a sequence of input data blocks assigned thereto by the master process and to apply an application-specific map function to each input data block in the sequence to produce the intermediate data; and

store the intermediate data in memory of the interconnected processors; and

in each of the plurality of reduce processes;

receive a respective partition of the intermediate data from the memory of the interconnected processors; and

apply an application-specific reduce function to the respective partition of the intermediate data to produce output values; and

in a respective reduce process;

receive multiple distinct partitions of the intermediate data and

process the multiple partitions one at a time in succession; and

identify the respective reduce process as a reduce process that is delaying the data processing job while continuing to process intermediate data and, in response, reassign at least one of the multiple partitions, which has not yet been processed, to a second reduce process, including copying the intermediate data in the reassigned partition to the other reduce process.

21. The non-transitory computer readable storage medium of claim 20, wherein:

the instructions to produce the intermediate data include instructions to produce a plurality of blocks of intermediate data, wherein each block of intermediate data includes all of the intermediate data produced by applying the application-specific map function to a respective block of input data; and

the instructions to receive a respective partition of the intermediate data include instructions to receive a subset of the intermediate data in a first block of intermediate data that is associated with the respective partition while a second block of intermediate data is being produced, the second block of intermediate data including at least some intermediate data that is associated with the respective partition.

22. The non-transitory computer readable storage medium of claim 20, wherein the one or more programs further comprise instructions,

responsive to identifying the respective reduce process as a reduce process that is delaying the data processing job, to divide the intermediate data in a partition that is assigned to the respective reduce process into a plurality of subpartitions and assign each subpartition to a reduce process that is not the respective reduce process.

23. The non-transitory computer readable storage medium of claim 20, wherein the instructions to apply an application-specific reduce function to the respective partition of the intermediate data to produce output values include instructions to:

while continuing to receive a respective partition of the intermediate data;

store at least a subset of the intermediate data of the respective partition in memory associated with the reduce process;

while the intermediate data is stored in the memory associated with the reduce process, apply an application-specific combiner function to produce combined intermediate data values; and

apply the application-specific reduce function to the combined intermediate data values to produce output values.

26. The non-transitory computer readable storage medium of claim 20, wherein receiving the respective partition of the intermediate data from the memory of the interconnected processors occurs while the map processes that produced the received intermediate data continue to process input data blocks.