Persistent shuffle system

1. A computer-implemented method performed by a data processing apparatus, the method comprising:

responsive to receiving a request to perform a shuffle operation on a data stream as part of a map reduce operation, the shuffle operation being an operation that groups keyed records in the data stream by key:

receiving at least a portion of the data stream, the data stream including a plurality of records, each record including a key, and the data stream received from one or more mappers of the map reduce operation;

during the shuffle operation, storing each of the plurality of records in a persistent storage location from a set of two or more persistent storage locations, each persistent storage location assigned to a key range corresponding to keys included in the plurality of records, the persistent storage location for each record being selected based at least in part on the key for the particular record that is within the key range assigned to the persistent storage location;

receiving a request, including a range of keys, from a reducer of the map reduce operation for a subset of the plurality of records stored in one or more of the persistent storage locations;

upon receiving the request from the reducer of the map reduce operation, providing the subset of the plurality of records included in the range of keys from the persistent storage locations to the reducer;

receiving a second request, including the range of keys, from a second reducer for the subset of the plurality of records stored in the one or more of the persistent storage locations; and

upon receiving the second request from the second reducer, providing the subset of the plurality of records included in the range of keys from the persistent storage locations to the second reducer.

2. The method of claim 1 , further comprising sorting the plurality of records by key during the shuffle operation.

3. The method of claim 1 , further comprising receiving a data consumption mode selection from the reducer, the data consumption mode indicating a manner in which the subset of the plurality of records including the range of keys should be provided to the reducer.

4. The method of claim 3 , wherein receiving the data consumption mode selection from the reducer includes receiving a streaming mode selection indicating that the plurality of records including the range of keys should be provided to the reducer immediately.

5. The method of claim 3 , wherein the receiving the data consumption mode selection from the reducer includes receiving a sharded mode selection indicating that the plurality of records including the range of keys should be provided to the reducer only after being stored in the persistent storage locations.

6. The method of claim 3 , wherein the receiving the data consumption mode selection from the reducer includes receiving a sorted mode selection indicating that the plurality of records including the range of keys should be provided to the reducer grouped and ordered by key.

7. The method of claim 1 , further comprising storing metadata associated with the shuffle operation in the persistent storage locations, wherein the metadata includes indexes associated with the plurality of records.

8. The method of claim 1 , further comprising:

receiving a request to delete the shuffle operation; and

upon receiving the request to delete the shuffle operation, deleting each of the plurality of records from the persistent storage locations.

9. The method of claim 1 , wherein storing each of the plurality of records in a persistent storage location includes appending the particular record to an end of a sequential file stored in the persistent storage location.

10. The method of claim 9 , further comprising tracking one or more next read locations in the sequential file, each of the one or more next read locations indicating a next record to deliver to one of one or more reducers.

11. The method of claim 1 , comprising:

receiving, from a reducer and during the shuffle operation, a request that indicates a first subset of keys of the plurality of records of the data stream that are already stored in the persistent storage locations and a second subset of keys of the plurality of records of the data stream that are not yet stored in the persistent locations; and

in response to receiving, from the reducer and during the shuffle operation, the request that indicates the subset of keys of the plurality of records of the data stream that are already stored in the persistent storage locations and the subset of keys of the plurality of records of the data stream that are not yet stored in the persistent storage locations, providing to the reducer the records that are already stored from the persistent storage locations and providing to the reducer the records that are not yet in the persistent storage locations before storing the records in the persistent storage locations.

12. A tangible, non-transitory computer-readable medium encoded with instructions for causing one or more processors to perform operations comprising:

responsive to receiving a request to perform a shuffle operation on a data stream as part of a map reduce operation, the shuffle operation being an operation that groups keyed records in the data stream by key:

receiving at least a portion of the data stream, the data stream including a plurality of records, each record including a key, and the data stream received from one or more mappers of the map reduce operation;

during the shuffle operation, storing each of the plurality of records in a persistent storage location from a set of two or more persistent storage locations, each persistent storage location assigned to a key range corresponding to keys included in the plurality of records, the persistent storage location for each record being selected based at least in part on the key for the particular record that is within the key range assigned to the persistent storage location;

receiving a request, including a range of keys, from a reducer of the map reduce operation for a subset of the plurality of records stored in one or more of the persistent storage locations;

upon receiving the request from the reducer of the map reduce operation, providing the subset of the plurality of records included in the range of keys from the persistent storage locations to the reducer;

receiving a second request, including the range of keys, from a second reducer for the subset of the plurality of records stored in the one or more of the persistent storage locations; and

upon receiving the second request from the second reducer, providing the subset of the plurality of records included in the range of keys from the persistent storage locations to the second reducer.

13. The computer-readable medium of claim 12 , further comprising sorting the plurality of records by key during the shuffle operation.

14. The computer-readable medium of claim 12 , further comprising receiving a data consumption mode selection from the reducer, the data consumption mode indicating a manner in which the subset of the plurality of records including the range of keys should be provided to the reducer.

15. The computer-readable medium of claim 14 , wherein receiving the data consumption mode selection from the reducer includes receiving a streaming mode selection indicating that the plurality of records including the range of keys should be provided to the reducer immediately.

16. The computer-readable medium of claim 14 , wherein the receiving the data consumption mode selection from the reducer includes receiving a sharded mode selection indicating that the plurality of records including the range of keys should be provided to the reducer only after being stored in the persistent storage locations.

17. The computer-readable medium of claim 14 , wherein the receiving the data consumption mode selection from the reducer includes receiving a sorted mode selection indicating that the plurality of records including the range of keys should be provided to the reducer grouped and ordered by key.

18. The computer-readable medium of claim 12 , the operations further comprising storing metadata associated with the shuffle operation in the one or more persistent storage locations, wherein the metadata includes indexes associated with the plurality of records.

19. The computer-readable medium of claim 12 , the operations further comprising:

receiving a request to delete the shuffle operation; and

upon receiving the request to delete the shuffle operation, deleting each of the plurality of records from the one or more persistent storage locations.

20. A system comprising:

memory for storing data; and

one or more processors operable to perform operations comprising:

responsive to receiving a request to perform a shuffle operation on a data stream as part of a map reduce operation, the shuffle operation being an operation that groups keyed records in the data stream by key:

receiving at least a portion of the data stream, the data stream including a plurality of records, each record including a key, and the data stream received from one or more mappers of the map reduce operation;

during the shuffle operation, storing each of the plurality of records in a persistent storage location from a set of two or more persistent storage locations, each persistent storage location assigned to a key range corresponding to keys included in the plurality of records, the persistent storage location for each record being selected based at least in part on the key for the particular record that is within the key range assigned to the persistent storage location;

receiving a request, including a range of keys, from a reducer of the map reduce operation for a subset of the plurality of records stored in one or more of the persistent storage locations;

upon receiving the request from the reducer of the map reduce operation, providing the subset of the plurality of records included in the range of keys from the persistent storage locations to the reducer;

receiving a second request, including the range of keys, from a second reducer for the subset of the plurality of records stored in the one or more of the persistent storage locations; and

upon receiving the second request from the second reducer, providing the subset of the plurality of records included in the range of keys from the persistent storage locations to the second reducer.