Opportunistic job processing

1. A method performed by one or more processors, the method comprising:

receiving, at a global-level manager, a work order referencing at least one binary executable file and one or more input files received from a client, the one or more input files including multiple input resources;

identifying a first type of input resource in the one or more input files and a second type of input resource in the one or more input files, the second type of input resource being smaller than the first type of input resource;

splitting the first type of input resource into a plurality of first-type input shards;

splitting the second type of input resource into a plurality of second-type input shards;

associating the plurality of second-type input shards with each of the first-type input shards;

for each first-type input shard, identifying a corresponding single cluster out of a plurality of clusters, each cluster comprising computing resources capable of executing the binary executable file to process first-type input and second-type input, wherein at least some of the computing resources are underutilized; and

distributing, to each of the identified single clusters by the global-level manager, the corresponding first-type input shards, the first-type input shards' associated second-type input shards, and the binary executable file such that the first-type input shard and the second type-input shard are processed by the binary executable file using underutilized computing resources in the cluster.

2. The method of claim 1 , the method further comprising:

receiving, at the single cluster, the first-type input shards, the associated second-type input shards, and the binary executable file; and

launching, for each second-type input shard, a worker to generate an output shard by executing the binary executable file on the first-type input shard and the second-type input shard, the worker executing using underutilized computing resources in the cluster.

3. The method of claim 1 , the method further comprising receiving, from each of the single clusters, output shards associated with the work order.

4. The method of claim 3 , the method further comprising generating, from the output shards, an output file for the work order.

5. The method of claim 4 , wherein the work order is received from a client system and wherein the method further comprises transmitting the output file to the client system.

6. The method of claim 1 , wherein the first type of input resource is a database and the second type of input is a query.

7. The method of claim 1 , wherein distributing, by the global-level manager, the first-type input shards, the associated second-type input shards, and the binary executable file to a single cluster out of a plurality of clusters comprises:

storing the first-type input shards, the associated second-type input shards, and the binary executable file in a central storage; and

distributing references to the first-type input shards, the associated second-type input shards, and the binary executable file to a single cluster out of a plurality of clusters such that the single cluster requests the first-type input shards, the associated second-type input shards, and the binary executable file from the central storage.

8. A computer system comprising:

computing hardware running a global-level manager configured to:

receive, from a client system, a work order referencing at least one binary executable file and one or more input files, the one or more input files including multiple input resources;

identify a first type of input resource in the one or more input files and a second type of input resource in the one or more input files, the second type of input resource being smaller than the first type of input resource;

split the first type of input resource into a plurality of first-type input shards;

split the second type of input resource into a plurality of second-type input shards;

associate the plurality of second-type input shards with each of the first-type input shards;

for each first-type input shard, identify a corresponding single cluster out of a plurality of clusters; and

distribute, to each of the identified single clusters, the corresponding first-type input shards, the first-type input shards' associated second-type input shards, and the binary executable file such that the first-type input shard and the second type-input shard are processed by the binary executable file using underutilized computing resources in the cluster;

a plurality of clusters, each cluster comprising computing resources capable of executing the binary executable file to process first-type input and second-type input, wherein at least some of the computing resources are underutilized; and

a client system configured to:

send, to the global-level manager, the work order.

9. The system of claim 8 , the system further comprising the single cluster configured to:

receive the first-type input shards, the associated second-type input shards, and

the binary executable file; and

launching, for each second-type input shard, a worker to generate an output shard by executing the binary executable file on the first-type input shard and the second-type input shard, the worker executing using underutilized computing resources in the cluster.

10. The system of claim 8 , wherein the global-level manager is further configured to receive, from each of the single clusters, output shards associated with the work order.

11. The system of claim 10 , wherein the global-level manager is further configured to generate, from the output shards, an output file for the work order.

12. The system of claim 11 , wherein the global-level manager is further configured to transmit the output file to the client system.

13. The system of claim 8 , wherein the first type of input resource is a database and the second type of input is a query.

14. The system of claim 8 , wherein the computer system further comprises a central storage; and

wherein, to distribute the first-type input shards, the associated second-type input shards, and the binary executable file to a single cluster out of a plurality of clusters, the global-level manager is configured to:

store the first-type input shards, the associated second-type input shards, and the binary executable file in the central storage; and

distribute references to the first-type input shards, the associated second-type input shards, and the binary executable file to a single cluster out of a plurality of clusters such that the single cluster requests the first-type input shards, the associated second-type input shards, and the binary executable file from the central storage.

15. A non-transitory tangible computer readable storage medium storing a computer program, the program comprising instructions that, when executed by one or more processing devices, cause the one or more processing devices to perform operations comprising:

receiving, at a global-level manager, a work order referencing at least one binary executable file and one or more input files received from a client, the one or more input files including multiple input resources;

identifying a first type of input resource in the one or more input files and a second type of input resource in the one or more input files, the second type of input resource being smaller than the first type of input resource;

splitting the first type of input resource into a plurality of first-type input shards;

splitting the second type of input resource into a plurality of second-type input shards;

associating the plurality of second-type input shards with each of the first-type input shards;

for each first-type input shard, identifying a corresponding single cluster out of a plurality of clusters, each cluster comprising computing resources capable of executing the binary executable file to process first-type input and second-type input, wherein at least some of the computing resources are underutilized; and

distributing, to each of the identified single clusters by the global-level manager, the corresponding first-type input shards, the first-type input shards' associated second-type input shards, and the binary executable file such that the first-type input shard and the second type-input shard are processed by the binary executable file using underutilized computing resources in the cluster.

16. The computer readable storage medium of claim 15 , the operations comprising:

receiving, at the single cluster, the first-type input shards, the associated second-type input shards, and the binary executable file; and

launching, for each second-type input shard, a worker to generate an output shard by executing the binary executable file on the first-type input shard and the second-type input shard, the worker executing using underutilized computing resources in the cluster.

17. The computer readable storage medium of claim 15 , the operations further comprising receiving, from each of the single clusters, output shards associated with the work order.

18. The computer readable storage medium of claim 17 , the operations further generating, from the output shards, an output file for the work order.

19. The computer readable storage medium of claim 18 , wherein the work order is received from a client system and wherein the operations further comprises transmitting the output file to the client system.

20. The computer readable storage medium of claim 15 , wherein the first type of input resource is a database and the second type of input is a query.

21. The computer readable storage medium of claim 15 , wherein distributing, by the global-level manager, the first-type input shards, the associated second-type input shards, and the binary executable file to a single cluster out of a plurality of clusters comprises:

storing the first-type input shards, the associated second-type input shards, and the binary executable file in a central storage; and

distributing references to the first-type input shards, the associated second-type input shards, and the binary executable file to a single cluster out of a plurality of clusters such that the single cluster requests the first-type input shards, the associated second-type input shards, and the binary executable file from the central storage.