TECHNICAL FIELD

This disclosure relates to parallel processing of data.

BACKGROUND

Large-scale data processing may include parallel processing, which generally involves performing some operation over each element of a large data set. The various operations may be chained together in a data-parallel pipeline to create an efficient mechanism for processing a data set.

SUMMARY

In one aspect, a system includes one or more processing devices and one or more storage devices. The storage devices store instructions that, when executed by the one or more processing devices, implement an application, an evaluator, an optimizer, and an executor. The application includes a data parallel pipeline. The data parallel pipeline specifies multiple parallel data objects that contain multiple elements and multiple parallel operations that operate on the parallel data objects. The evaluator is configured, based on the data parallel pipeline, to generate a dataflow graph of deferred parallel data objects and deferred parallel operations corresponding to the data parallel pipeline.

Deferred parallel data objects, for example, can be data structures that include a pointer to the parallel data operation that operates on the parallel data objects, rather than the elements stored in the parallel data object. Deferred parallel operations, for example, can be data structures that include a pointer to a parallel data object that is an input to the deferred parallel operation, a pointer to a deferred parallel object that is an output of the deferred parallel operation, and a function to be (but has not yet been) performed on the input object.

The optimizer is configured to apply one or more graph transformations to the dataflow graph to generate a revised dataflow graph that includes one or more of the deferred parallel data objects and deferred, combined parallel data operations. The executor configured to execute the deferred, combined parallel operations to produce materialized parallel data objects corresponding to the deferred parallel data objects. Materialized parallel data objects, for example, can be data structures that include the data or elements of the parallel data object.

Implementations of this aspect may include one or more of the following features. For example, the deferred, combined parallel data operations may include at least one generalized mapreduce operation. The generalized mapreduce operation may include multiple, parallel map operations and multiple, parallel reduce operations and be translatable to a single mapreduce operation that includes a single map function to implement the multiple, parallel map operations and a single reduce function to implement the multiple, parallel reduce operations. To execute the generalized mapreduce operation, the executor may be configured to translate the combined mapreduce operation to the single mapreduce operation and execute the single mapreduce operation. To execute the single mapreduce operation, the executor may be configured to determine whether to execute the single mapreduce operation as a local, sequential operation or a remote, parallel operation. To translate the generalized mapreduce operation to the single mapreduce operation, the executor may be configured to generate a map function that includes the multiple map operations and a reducer function that includes the multiple reducer operations.

The multiple parallel data objects may be first class objects of a host programming language.

The pipeline further may include a single data object that contains a single element and the dataflow graph includes a corresponding deferred single data object. At least one of the multiple parallel operations in the pipeline may operate on the single data object and one of the multiple parallel data objects and the dataflow graph may include a corresponding deferred parallel operation that operates on a deferred single data object and a deferred parallel data object.

The executor may be configured to cache one or more results of the execution of the deferred, combined parallel operations for use in a future execution of the data parallel pipeline.

In another aspect, a method includes executing an application that includes a data parallel pipeline. The data parallel pipeline specifies multiple parallel data objects that contain multiple elements and multiple parallel operations that operate on the parallel data objects. The method further includes generating, based on the data parallel pipeline, a dataflow graph of deferred parallel data objects and deferred parallel operations corresponding to the data parallel pipeline. Deferred parallel data objects, for example, can be data structures that include a pointer to the parallel data operation that operates on the parallel data objects, rather than the elements stored in the parallel data object. Deferred parallel operations, for example, can be data structures that include a pointer to a parallel data object that is an input to the deferred parallel operation, a pointer to a deferred parallel object that is an output of the deferred parallel operation, and a function to be (but has not yet been) performed on the input object.

The method also includes applying one or more graph transformations to the dataflow graph to generate a revised dataflow graph that includes one or more of the deferred parallel data objects and deferred, combined parallel data operations. In addition, the method includes executing the deferred, combined parallel operations to produce materialized parallel data objects corresponding to the deferred parallel data objects. Materialized parallel data objects, for example, can be data structures that include the data or elements of the parallel data object.

Implementations of this aspect may include one or more of the following features. For example, the deferred, combined parallel data operations may include at least one generalized mapreduce operation. The generalized mapreduce operation may include multiple, parallel map operations and multiple, parallel reduce operations and be translatable to a single mapreduce operation that includes a single map function to implement the multiple, parallel map operations and a single reduce function to implement the multiple, parallel reduce operations. Executing the generalized mapreduce operation may include translating the combined mapreduce operation to the single mapreduce operation and executing the single mapreduce operation. Executing the single mapreduce operation may include determining whether to execute the single mapreduce operation as a local, sequential operation or a remote, parallel operation. Translating the generalized mapreduce operation to the single mapreduce operation may include generating a map function that includes the multiple map operations and a reducer function that includes the multiple reducer operations.

The multiple parallel data objects may be first class objects of a host programming language.

The pipeline further may include a single data object that contains a single element and the dataflow graph includes a corresponding deferred single data object. At least one of the multiple parallel operations in the pipeline may operate on the single data object and one of the multiple parallel data objects and the dataflow graph may include a corresponding deferred parallel operation that operates on a deferred single data object and a deferred parallel data object.

The method may include caching one or more results of the execution of the deferred, combined parallel operations for use in a future execution of the data parallel pipeline.

In another aspect, a system includes one or more processing devices and one or more storage devices. The storage devices store instructions that, when executed by the one or more processing devices, implement an executor. The executor is configured to access a dataflow graph that includes deferred parallel data objects and deferred, combined parallel data operations. Deferred parallel data objects, for example, can be data structures that include a pointer to the parallel data operation that operates on the parallel data objects, rather than the elements stored in the parallel data object. Deferred parallel operations, for example, can be data structures that include a pointer to a parallel data object that is an input to the deferred parallel operation, a pointer to a deferred parallel object that is an output of the deferred parallel operation, and a function to be (but has not yet been) performed on the input object.

The executor is configured to execute the deferred, combined parallel operations to produce materialized parallel data objects corresponding to the deferred parallel data objects. Materialized parallel data objects, for example, can be data structures that include the data or elements of the parallel data object. For at least one of the deferred, combined parallel operations, the executor is configured to execute the at least one deferred, combined parallel operation by determining an estimated size of data associated with the at least one deferred, combined parallel operation; determining whether the estimated size exceeds a threshold size; if the estimated size is below a threshold size, execute the at least one deferred, combined parallel operation as a local, sequential operation; and if the estimated size exceeds a threshold size, execute the at least one deferred, combined parallel operation as remote, parallel operation.

Implementations of this aspect may include one or more of the following features. For example, the data associated with the at least one deferred, combined parallel operation may include one or more of input data for the at least one deferred, combined parallel operation, intermediary data produced by the at least one deferred, combined parallel operation, or output data produced by the at least one deferred, combined parallel operation. The at least one deferred, combined parallel data operation may be a generalized mapreduce operation. The generalized mapreduce operation may include multiple, parallel map operations and multiple, parallel reduce operations and be translatable to a single mapreduce operation that includes a single map function to implement the multiple, parallel map operations and a single reduce function to implement the multiple, parallel reduce operations. To execute the generalized mapreduce operation, the executor may be configured to translate the combined mapreduce operation to the single mapreduce operation and execute the single mapreduce operation. To execute the single mapreduce operation as a remote, parallel operation, the executor may be configured to cause the single mapreduce operation to be copied and executed on multiple, different processing modules in a datacenter. To translate the generalized mapreduce operation to the single mapreduce operation, the executor may be configured to generate a map function that includes the multiple map operations and a reducer function that includes the multiple reducer operations. To determine the estimated size, the executor may be configured to access annotations in the dataflow graph that reflect an estimate of the size of the data associated with the at least one deferred, combined parallel operation.

In another aspect, a method includes accessing a dataflow graph that includes deferred parallel data objects and deferred, combined parallel data operations. Deferred parallel data objects, for example, can be data structures that include a pointer to the parallel data operation that operates on the parallel data objects, rather than the elements stored in the parallel data object. Deferred parallel operations, for example, can be data structures that include a pointer to a parallel data object that is an input to the deferred parallel operation, a pointer to a deferred parallel object that is an output of the deferred parallel operation, and a function to be (but has not yet been) performed on the input object.

The method also includes executing the deferred, combined parallel operations to produce materialized parallel data objects corresponding to the deferred parallel data objects. Materialized parallel data objects, for example, can be data structures that include the data or elements of the parallel data object. For at least one of the deferred, combined parallel operations, executing the at least one deferred, combined parallel operation includes determining an estimated size of data associated with the at least one deferred, combined parallel operation; determining whether the estimated size exceeds a threshold size; if the estimated size is below a threshold size, execute the at least one deferred, combined parallel operation as a local, sequential operation; and if the estimated size exceeds a threshold size, execute the at least one deferred, combined parallel operation as remote, parallel operation.

Implementations of this aspect may include one or more of the following features. For example, the data associated with the at least one deferred, combined parallel operation may include one or more of input data for the at least one deferred, combined parallel operation, intermediary data produced by the at least one deferred, combined parallel operation, or output data produced by the at least one deferred, combined parallel operation. The at least one deferred, combined parallel data operation may be a generalized mapreduce operation. The generalized mapreduce operation may include multiple, parallel map operations and multiple, parallel reduce operations and be translatable to a single mapreduce operation that includes a single map function to implement the multiple, parallel map operations and a single reduce function to implement the multiple, parallel reduce operations. Executing the generalized mapreduce operation may include translating the combined mapreduce operation to the single mapreduce operation and executing the single mapreduce operation. Executing the single mapreduce operation as a remote, parallel operation may include causing the single mapreduce operation to be copied and executed on multiple, different processing modules in a datacenter. Translating the generalized mapreduce operation to the single mapreduce operation may include generating a map function that includes the multiple map operations and a reducer function that includes the multiple reducer operations. Determining the estimated size may include accessing annotations in the dataflow graph that reflect an estimate of the size of the data associated with the at least one deferred, combined parallel operation.

In one aspect, a system includes one or more processing devices and one or more storage devices. The storage devices store instructions that, when executed by the one or more processing devices, implement an executor. The executor is configured to access a dataflow graph that includes deferred parallel data objects and deferred, combined parallel data operations. Deferred parallel data objects, for example, can be data structures that include a pointer to the parallel data operation that operates on the parallel data objects, rather than the elements stored in the parallel data object. Deferred parallel operations, for example, can be data structures that include a pointer to a parallel data object that is an input to the deferred parallel operation, a pointer to a deferred parallel object that is an output of the deferred parallel operation, and a function to be (but has not yet been) performed on the input object.

At least one of the deferred, combined parallel data operation is a generalized mapreduce operation. The generalized mapreduce operation includes multiple, parallel map operations and multiple, parallel reduce operations and is translatable to a single mapreduce operation that includes a single map function to implement the multiple, parallel map operations and a single reduce function to implement the multiple, parallel reduce operations.

The executor is further configured to execute the deferred, combined parallel operations to produce materialized parallel data objects corresponding to the deferred parallel data objects. Materialized parallel data objects, for example, can be data structures that include the data or elements of the parallel data object. To execute the generalized mapreduce operation, the executor is configured to translate the combined mapreduce operation to the single mapreduce operation and execute the single mapreduce operation.

Implementations of this aspect may include one or more of the following features. For example, to translate the generalized mapreduce operation to the single mapreduce operation, the executor may be configured to generate a map function that includes the multiple map operations and a reduce function that includes the multiple reduce operations.

The executor may be configured to execute the single mapreduce operation as a remote, parallel operation. To execute the single mapreduce operation as a remote, parallel operation, the executor may be configured to cause the single mapreduce operation to be executed by multiple worker processes on multiple, different processing modules. To cause the single map reduce operation to be executed by multiple worker processes, the executor may be configured to cause multiple map worker processes to be invoked for each of the map operations, where each of the multiple map worker processes is assigned an index number. Each of the map worker processes may be configured to receive the map function that implements the multiple map operations, one or more inputs associated with one of the map operations, and the map worker process' associated index, select the map operation associated with the inputs based on the map worker process' associated index, and invoke the selected map operation on the one or more inputs.

To cause the single map reduce operation to be executed by multiple worker processes, the executor may be configured to cause multiple reduce worker processes to be invoked for each of the reduce operations, where each of the multiple reduce worker processes is assigned an index number. Each of the multiple reduce worker processes may be configured to receive the reduce function that implements the multiple reduce operations, one or more inputs associated with one of the reduce operations, and the reduce worker process' associated index, select the reduce operation associated with the inputs based on the worker process' associated index, and invoke the selected reduce operation on the one or more inputs.

In another aspect, a method includes accessing a dataflow graph that includes deferred parallel data objects and deferred, combined parallel data operations. Deferred parallel data objects, for example, can be data structures that include a pointer to the parallel data operation that operates on the parallel data objects, rather than the elements stored in the parallel data object. Deferred parallel operations, for example, can be data structures that include a pointer to a parallel data object that is an input to the deferred parallel operation, a pointer to a deferred parallel object that is an output of the deferred parallel operation, and a function to be (but has not yet.

At least one of the deferred, combined parallel data operation is a generalized mapreduce operation. The generalized mapreduce operation includes multiple, parallel map operations and multiple, parallel reduce operations and being translatable to a single mapreduce operation that includes a single map function to implement the multiple, parallel map operations and a single reduce function to implement the multiple, parallel reduce operations; and

The method also includes executing the deferred, combined parallel operations to produce materialized parallel data objects corresponding to the deferred parallel data objects. Materialized parallel data objects, for example, can be data structures that include the data or elements of the parallel data object. Executing the generalized mapreduce operation includes translating the combined mapreduce operation to the single mapreduce operation and executing the single mapreduce operation.

Implementations of this aspect may include one or more of the following features. For example, translating the generalized mapreduce operation to the single mapreduce operation may include generating a map function that includes the multiple map operations and a reduce function that includes the multiple reduce operations. Executing the single mapreduce operation may include executing the single mapreduce operation as a remote, parallel operation. Executing the single mapreduce operation as a remote, parallel operation may include causing the single mapreduce operation to be executed by multiple worker processes on multiple, different processing modules.

Causing the single map reduce operation to be executed by multiple worker processes may include causing the multiple map worker processes to be invoked for each of the map operations, where each of the multiple map worker processes is assigned an index number. Each of the map worker processes may be configured to receive the map function that implements the multiple map operations, one or more inputs associated with one of the map operations, and the map worker process' associated index, select the map operation associated with the inputs based on the map worker process' associated index, and invoke the selected map operation on the one or more inputs.

Causing the single map reduce operation to be executed by multiple worker processes may include causing multiple reduce worker processes to be invoked for each of the reduce operations, where each of the multiple reduce worker processes is assigned an index number. Each of the multiple reduce worker processes may be configured to receive the reduce function that implements the multiple reduce operations, one or more inputs associated with one of the reduce operations, and the reduce worker process' associated index, select the reduce operation associated with the inputs based on the worker process' associated index, and invoke the selected reduce operation on the one or more inputs.

