INFRASTRUCTURE FOR PARALLEL PROGRAMMING OF CLUSTERS OF MACHINES

• US 20090089560A1
• Filed: 11/07/2008
• Published: 04/02/2009
• Est. Priority Date: 10/01/2007
• Status: Active Grant

• Current Assignee: Accenture Global Services Limited (Accenture PLC)
• Original Assignee: Accenture PLC
• Inventors: Liu, Huan, Orban, Dan
• Granted Patent: US 7,970,872 B2
• Time in Patent Office: Days
• US Class Current: 712/226
• CPC Class Codes: G06F 8/45 Exploiting coarse grain parallelism in compilation, i.e. parallelism between groups of instructions
• 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

GridBatch provides an infrastructure framework that hides the complexities and burdens of developing logic and programming application that implement detail parallelized computations from programmers. A programmer may use GridBatch to implement parallelized computational operations that minimize network bandwidth requirements, and efficiently partition and coordinate computational processing in a multiprocessor configuration. GridBatch provides an effective and lightweight approach to rapidly build parallelized applications using economically viable multiprocessor configurations that achieve the highest performance results.

Claims

1. A product comprising:

a machine readable medium;

first operator logic stored on the medium and operable to;

implement a first data processing operation in parallel over multiple processing nodes, the first data processing operation customized with a first user-defined function executed on the multiple processing nodes; and

second operator logic stored on the medium and operable to;

implement a second data processing operation in parallel over the multiple processing nodes, the second data processing operation customized with a second user-defined function executed on the multiple processing nodes;

where the first or second operator logic comprises block operator logic;

where the first or second user-defined function comprises a user-defined block function; and

where the block operator logic is operable to apply the user-defined block function to records of a first vector.

2. The product of claim 1, further comprising:

file system manager logic stored on the medium, and operable to;

assign vector chunks of the first vector among the multiple processing nodes according to a user-defined hash function.

3. The product of claim 2, where the file system manager logic is further operable to:

provide vector chunk node location information for the vector chunks to a job scheduler.

4. The product of claim 2, where the file system manager logic is further operable to:

reshuffle the vector chunks.

5. The product of claim 2, where the file system manager logic is further operable to:

maintain a mapping of chunk IDs to the multiple processing nodes that identifies each chunk ID data node assignment; and

reshuffle the vector chunks when the mapping changes.

6. The product of claim 1, where:

the first or second operator logic comprises join operator logic;

the first or second user-defined function comprises a user-defined join function; and

where the join operator logic is operable to invoke the user-defined join function on a first matching record in the first vector and a second matching record in a second vector distributed among the multiple processing nodes when a join index field present in the first vector and the second vector matches for the first matching record and the second matching record, to obtain a join result.

7. The product of claim 6, further comprising:

master node logic stored on the medium and operable to;

receive a join function call; and

initiate spawning of joining tasks locally among the multiple processing nodes, each joining task operable to selectively initiate execution of the user-defined join function.

8. The product of claim 1, where:

the first or second operator logic comprises recurse operator logic;

the first or second user-defined function comprises a user-defined recurse function; and

where the recurse operator logic is operable to invoke the user-defined recurse function starting over the vector chunks locally on the multiple processing nodes to produce intermediate results, communicate a subset of the intermediate results to a subset of the multiple processing nodes, and iterate;

invocation of the user-defined recurse function on the intermediate results to

produce increasingly fewer intermediate results; and

communication of a subset of the increasingly fewer intermediate results to an increasingly smaller subset of the multiple processing nodes;

until a final recurse result is obtained over the first vector on a final node in the first set of nodes.

9. The product of claim 8, further comprising:

master node logic stored on the medium and operable to;

receive a recurse function call; and

initiate spawning of recurse operation tasks locally among the multiple processing nodes, each recurse operation task operable to selectively initiate execution of the user-defined recurse function to the vector chunks.

10. The product of claim 1, where:

the first or second operator logic comprises convolution operator logic;

the first or second user-defined function comprises a user-defined convolution function; and

where the convolution operator logic is operable to invoke the user-defined convolution function for each record in the first vector on every record in a second vector, to obtain a convolution function result.

11. The product of claim 10, further comprising:

master node logic stored on the medium and operable to;

receive a convolution function call; and

initiate spawning of convolution operation tasks locally among the multiple processing nodes, each convolution operation task operable to selectively initiate execution of the user-defined convolution function.

12. The product of claim 1, where:

the first or second operator logic comprises distribute operator logic;

the first or second user-defined function comprises a user-defined partition function; and

where the distribute operator logic is operable to redistribute, according to the user-defined partition function, a first vector previously distributed as first vector chunks among the multiple processing nodes, to obtain redistributed vector chunks of the first vector redistributed among the multiple processing nodes.

13. The product of claim 1, where:

the first or second operator logic comprises map operator logic;

the first or second user-defined function comprises a user-defined map function; and

where the map operator logic is operable to apply the user-defined map function to the records of a second vector distributed among the multiple processing nodes.

14. A method for processing data in parallel comprising:

initiating execution of a first data processing operation in parallel over multiple processing nodes, the first data processing operation customized with a first user-defined function executed on the multiple processing nodes; and

initiating execution of a second data processing operation in parallel over the multiple processing nodes, the second data processing operation customized with a second user-defined function executed on the multiple processing nodes;

where the first or second data processing operation comprises block operator logic;

where the first or second user-defined function comprises a user-defined block function; and

where the block operator logic applies the user-defined block function to records of a first vector.

15. The method of claim 14, further comprising:

assigning vector chunks of the first vector among the multiple processing nodes according to a user-defined hash function.

16. The method of claim 14, further comprising:

providing vector chunk node location information for the vector chunks to a job scheduler.

17. The method of claim 14, further comprising:

reshuffling the vector chunks.

18. The method of claim 14, where:

the first or second data operations processing operation comprises join operator logic;

the first or second user-defined function comprises a user-defined join function; and

where the join operator logic invokes the user-defined join function on a first matching record in the first vector and a second matching record in a second vector distributed among the multiple processing nodes when a join index field present in the first vector and the second vector matches for the first matching record and the second matching record, to obtain a join result.

19. The method of claim 18, further comprising:

receiving a join function call; and

initiating spawning of joining tasks locally among the multiple processing nodes, each joining task operable to selectively initiate execution of the user-defined join function.

20. The method of claim 14, where:

the first or second data operations processing operation comprises recurse operator logic;

the first or second user-defined function comprises a user-defined recurse function; and

where the recurse operator logic invokes the user-defined recurse function starting over vector chunks locally on the multiple processing nodes to produce intermediate results, communicate a subset of the intermediate results to a subset of the multiple processing nodes, and iterate;

invocation of the user-defined recurse function on the intermediate results to produce increasingly fewer intermediate results; and

communication of a subset of the increasingly fewer intermediate results to an increasingly smaller subset of the multiple processing nodes;

until a final recurse result is obtained over the first vector on a final node in the first set of nodes.

21. The method of claim 20, further comprising:

receiving a recurse function call; and

initiating spawning of recurse operation tasks locally among the multiple processing nodes, each recurse operation task operable to selectively initiate execution of the user-defined recurse function to the vector chunks.

22. The method of claim 14, where:

the first or second data operations processing operation comprises convolution operator logic;

the first or second user-defined function comprises a user-defined convolution function; and

where the convolution operator logic invokes the user-defined convolution function for each record in the first vector on every record in a second vector, to obtain a convolution function result.

23. The method of claim 22, further comprising:

receiving a convolution function call; and

initiating spawning of convolution operation tasks locally among the multiple processing nodes, each convolution operation task operable to selectively initiate execution of the user-defined convolution function.

24. The method of claim 14, where:

the first or second data operations processing operation comprises distribute operator logic;

the first or second user-defined function comprises a user-defined partition function; and

where the distribute operator logic redistributes, according to the user-defined partition function, the first vector previously distributed as first vector chunks among the multiple processing nodes, to obtain redistributed vector chunks of the first vector redistributed among the multiple processing nodes.

25. The method of claim 14, where:

the first or second data operations processing operation comprises map operator logic;

the first or second user-defined function comprises a user-defined map function; and

where the map operator logic applies the user-defined map function to records of a vector distributed among the multiple processing nodes.