Propagating unobserved exceptions in distributed execution environments
The present invention extends to methods, systems, and computer program products for propagating unhandled exceptions in distributed execution environments, such as clusters. A job (e.g., a query) can include a series of computation steps that are executed on multiple compute nodes each processing parts of a distributed data set. Unhandled exceptions can be caught while computations are running on data partitions of different compute nodes. Unhandled exception objects can be stored in a serialized format in a compute node's local storage (or an alternate central location) along with auxiliary details such as the data partition being processed at the time. Stored serialized exception objects for a job can be harvested and aggregated in a single container object. The single container object can be passed back to the client.
1. At a distributed data processing system, the distributed data processing system including a client, a distributed data set, and a compute node cluster, the distributed data set partitioned into a plurality of data partitions, the compute node cluster including a plurality of compute nodes, a method for propagating unhandled exceptions from the plurality of compute nodes to the client, the method comprising:
an act of receiving a data processing plan from the client, the data processing plan defining how a data access request for specified data from within the distributed data set is to be executed across the plurality of compute nodes within the compute node cluster, the data processing plan including a plurality of different stages, each different stage defining a type of processing that is to be performed on data within the distributed data set;
an act of orchestrating the data processing plan on the compute node cluster by executing one or more of the plurality of different stages of the data processing plan, including for each stage:
an act of instructing one or more compute nodes to: (a) execute instructions against data within or derived from one or more data partitions of the distributed data set, (b) catch any encountered unhandled exceptions occurring during execution of the instructions against the data within or derived from one more data partitions, each unhandled exception indicating at least one of: a failure at a particular node or a failure at a particular stage of the data processing plan, and (c) store encountered unhandled exceptions along with corresponding auxiliary data in a designated storage location, the corresponding auxiliary data associating each of the unhandled exceptions with the data partition, from among the one or more data partitions, that lead to the unhandled exception;
an act of harvesting stored unhandled exceptions along with corresponding auxiliary data;
an act of aggregating the harvested unhandled exceptions along with corresponding auxiliary data into one or more container objects, the one or more container objects conveying details of compute node and stage failures that happened during orchestration of the data processing plan; and
an act of sending the one or more container objects to the client.
2. The method as recited in claim 1 , wherein the act of receiving a data processing plan from the client comprises an act of receiving a query job from the client, the query job including a query plan having a plurality of stages and including one or more user defined functions.
3. The method as recite in claim 1 , further comprising prior to orchestrating the data processing plan, an act of refining the data processing plan based on characteristics of the compute node cluster and/or based on characteristics of the data partitions.
4. The method as recited in claim 3 , wherein an act of refining the data processing plan comprises an act of refining the data processing plan based on one or more of: available compute nodes in the compute node cluster, number of input data partitions in the computer node cluster, locations of data partitions within the computer node cluster, and the network topology of the compute node cluster.
5. The method as recited in claim 1 , wherein the act of harvesting stored unhandled exceptions along with corresponding auxiliary data comprises an act of the client traversing file systems on each compute node that participated in the data processing plan.
6. The method as recited in claim 1 , wherein the act of harvesting stored unhandled exceptions along with corresponding auxiliary data comprises an act of a head node for the compute node cluster traversing file systems on each compute node that participated in the data processing plan.
7. The method as recited in claim 1 , wherein the act of harvesting stored unhandled exceptions along with corresponding auxiliary data comprises an act of orchestrating a follow up data processing plan to copy stored unhandled exceptions along with corresponding auxiliary data to a central location.
8. The method as recited in claim 1 , wherein the act of harvesting stored unhandled exceptions along with corresponding auxiliary data comprises an act of harvesting stored unhandled exceptions along with corresponding auxiliary data while the data processing plan is executing.
9. The method as recited in claim 1 , wherein further comprising an act of detecting interruption of the data processing plan, the data processing plan interrupted based on one or more unhandled exceptions; and
wherein the act of harvesting stored unhandled exceptions along with corresponding auxiliary data comprises and act of harvesting stored unhandled exceptions along with corresponding auxiliary data in response to detecting interruption of the data processing plan.
10. The method as recited in claim 1 , wherein the act of sending the one or more container objects to the client comprises an act of throwing an aggregate exception from client side Application Program Interfaces (“APIs”).
11. The method as recited in claim 1 , wherein the act of sending the one or more container objects to the client comprises an act of sending the one or more container objects through a notification API.
12. A computer program product for use in a distributed data processing system, the distributed data processing system including a client, a distributed data set, and a compute node cluster, the distributed data set partitioned in to a plurality of data partitions, the compute node cluster including a plurality of compute nodes, the computer program product for implementing a method for propagating unhandled exceptions from the plurality of compute nodes to the client, the computer program product comprising one or more computer storage devices having stored thereon computer-executable instructions that, when executed at a processor, cause the distributed data processing system to perform the method, including the following:
receive a data processing plan from the client, the data processing plan defining how a data access request for specified data from within the distributed data set is to be executed across the plurality of compute nodes within the compute node cluster, the data processing plan including a plurality of different stages, each different stage defining a type of processing that is to be performed on data within the distributed data set;
orchestrate the data processing plan on the compute node cluster by executing one or more of the plurality of different stages of the data processing plan, including for each stage:
instructing one or more compute nodes to: (a) execute instructions against data within or derived from one or more data partitions of the distributed data set, (b) catch any encountered unhandled exceptions occurring during execution of the instructions against the data within or derived from one more data partitions, each unhandled exception indicating at least one of: a failure at a particular node or a failure at a particular stage of the data processing plan, and (c) store encountered unhandled exceptions along with corresponding auxiliary data in a designated storage location, the corresponding auxiliary data associating each of the unhandled exceptions with the data partition, from among the one or more data partitions, that lead to the unhandled exception;
detect interruption of the data processing plan, the data processing plan interrupted based on the one or more of unhandled exceptions;
harvest the stored unhandled exceptions along with corresponding auxiliary data in response to interruption of the data processing plan;
aggregate the harvested unhandled exceptions along with corresponding auxiliary data into one or more container objects, the one or more container objects conveying details of compute node and stage failures that happened during orchestration of the data processing plan; and
send the one or more container objects to the client.
13. The computer program product as recited in claim 12 , wherein computer-executable instructions that, when executed, cause the distributed data processing system to receiving a data processing plan from the client comprise computer-executable instructions that, when executed, cause the distributed data processing system to receive a query job from the client, the query job including a query plan having a plurality of stages and including one or more user defined functions.
14. The computer program product as recited in claim 12 , further comprising computer-executable instructions that, when executed, cause the distributed data processing system to refine the data processing plan based on characteristics of the compute node cluster and/or based on characteristics of the data partitions.
15. The computer program product as recited in claim 12 , wherein computer-executable instructions that, when executed, cause the distributed data processing system to harvest stored unhandled exceptions along with corresponding auxiliary data comprise computer-executable instructions that, when executed, cause the distributed data processing system to receive a client traversal of file systems on each compute node that participated in the data processing plan.
16. The computer program product as recited in claim 12 , wherein computer-executable instructions that, when executed, cause the distributed data processing system to harvest stored unhandled exceptions along with corresponding auxiliary data comprise computer-executable instructions that, when executed, cause a head node for the compute node cluster to traverse file systems on each compute node that participated in the data processing plan.
17. The computer program product as recited in claim 12 , wherein computer-executable instructions that, when executed, cause the distributed data processing system to harvest stored unhandled exceptions along with corresponding auxiliary data comprise computer-executable instructions that, when executed, cause the distributed data processing system to orchestrate a follow up data processing plan to copy stored unhandled exceptions along with corresponding auxiliary data to a central location.
18. A distributed data processing system, the distributed data processing system comprising a client, a job manager, and a compute node cluster,
wherein the compute node cluster includes a plurality of computer nodes and data partitions, each compute node including one or more processors and system memory, each compute node configured to execute code based on received instructions from the job manager;
wherein the client includes an application and a client runtime, the application being configured to generate a query having a plurality of query operators and one or more user defined functions, the client runtime being configured to receive the query and formulate at query plan from the query, the query plan having plurality of stages, wherein the client is configured to combine the query plan and user defined functions into a query job;
wherein the job manager is configured to:
receive the query job and refine the query job into a refined execution plan for execution within the compute node cluster; and
orchestrate the refined execution plan on the compute node cluster by executing one or more of the plurality of different stages of the data processing plan, including for each stage:
instructing one or more compute nodes to: (a) execute instructions against data within or derived from one or more data partitions of a distributed data set, (b) catch any encountered unhandled exceptions occurring during execution of the instructions against the data within or derived from one more data partitions, each unhandled exception indicating at least one of: a failure at a particular node or a failure at a particular stage of the data processing plan, and (c) store encountered unhandled exceptions along with corresponding auxiliary data in a designated storage location, the corresponding auxiliary data associating each of the unhandled exceptions with the data partition, from among the one or more data partitions, that lead to the unhandled exception; and
wherein each of the one or more compute nodes runs vertices in accordance with the refined execution plan to (a) execute instructions against one or more data partitions, (b) catch any unhandled exceptions, and (c) serialize encountered unhandled exceptions along with corresponding auxiliary data in the designated storage location; and
wherein at least one vertex catches an unhandled exception, serializes the unhandled exception into a serialized unhandled exception including corresponding auxiliary data, and stores the serialized unhandled exception for harvesting.
19. The distributed data processing system of claim 18 , wherein designated locations for storing unhandled exceptions are vertex directories at the one or more compute nodes.
20. The distributed data processing system of claim 18 , wherein the application being configured to generate a query having a plurality of query operators comprises the application being configured to generate a LINQ to HPC query.