BACKGROUND

This description relates to managing a computing cluster.

One approach to data flow computation makes use of a graph-based representation in which computational components corresponding to nodes (vertices) of a graph are coupled by data flows corresponding to links (directed edges) of the graph (called a “dataflow graph”). A downstream component connected to an upstream component by a data flow link receives an ordered stream of input data elements and processes the input data elements in the received order, optionally generating one or more corresponding flows of output data elements. A system for executing such graph-based computations is described in prior U.S. Pat. No. 5,966,072, titled “EXECUTING COMPUTATIONS EXPRESSED AS GRAPHS,” incorporated herein by reference. In an implementation related to the approach described in that prior patent, each component is implemented as a process that is hosted on one of typically multiple computer servers. Each computer server may have multiple such component processes active at any one time, and an operating system (e.g., Unix) scheduler shares resources (e.g., processor time, and/or processor cores) among the components hosted on that server. In such an implementation, data flows between components may be implemented using data communication services of the operating system and data network connecting the servers (e.g., named pipes, TCP/IP sessions, etc.). A subset of the components generally serve as sources and/or sinks of data from the overall computation, for example, to and/or from data files, database tables, and external data flows. After the component processes and data flows are established, for example, by a coordinating process, data then flows through the overall computation system implementing the computation expressed as a graph generally governed by availability of input data at each component and scheduling of computing resources for each of the components. Parallelism can therefore be achieved at least by enabling different components to be executed in parallel by different processes (hosted on the same or different server computers or processor cores), where different components executing in parallel on different paths through a dataflow graph is referred to herein as component parallelism, and different components executing in parallel on different portion of the same path through a dataflow graph is referred to herein as pipeline parallelism.

Other forms of parallelism are also supported by such an approach. For example, an input data set may be partitioned, for example, according to a partition of values of a field in records of the data set, with each part being sent to a separate copy of a component that processes records of the data set. Such separate copies (or “instances”) of a component may be executed on separate server computers or separate processor cores of a server computer, thereby achieving what is referred to herein as data parallelism. The results of the separate components may be merged to again form a single data flow or data set. The number of computers or processor cores used to execute instances of the component would be designated by a developer at the time the dataflow graph is developed.

Various approaches may be used to improve efficiency of such an approach. For example, each instance of a component does not necessarily have to be hosted in its own operating system process, for example, using one operating system process to implement multiple components (e.g., components forming a connected subgraph of a larger graph).

At least some implementations of the approach described above suffer from limitations in relation to the efficiency of execution of the resulting processes on the underlying computer servers. For example, the limitations may be related to difficulty in reconfiguring a running instance of a graph to change a degree of data parallelism, to change to servers that host various components, and/or to balance load on different computation resources. Existing graph-based computation systems also suffer from slow startup times, often because too many processes are initiated unnecessarily, wasting large amounts of memory. Generally, processes start at the start-up of graph execution, and end when graph execution completes.

Other systems for distributing computation have been used in which an overall computation is divided into smaller parts, and the parts are distributed from one master computer server to various other (e.g., “slave”) computer servers, which each independently perform a computation, and which return their result to a master server. Some of such approaches are referred to as “grid computing.” However, such approaches generally rely on the independence of each computation, without providing a mechanism for passing data between the computation parts, or scheduling and/or sequencing execution of the parts, except via the master computer server that invokes those parts. Therefore, such approaches do not provide a direct and efficient solution to hosting computation involving interactions between multiple components.

Another approach for distributed computation on a large dataset makes use of a MapReduce framework, for example, as embodied in the Apache Hadoop® system. Generally, Hadoop has a distributed filesystem in which parts for each named file are distributed. A user specifies a computation in terms of two functions: a map function, which is executed on all the parts of the named inputs in a distributed manner, and a reduce function that is executed on parts of the output of the map function executions. The outputs of the map function executions are partitioned and stored in intermediate parts again in the distributed filesystem. The reduce function is then executed in a distributed manner to process the intermediate parts, yielding the result of the overall computation. Although computations that can be expressed in a MapReduce framework, and whose inputs and outputs are amendable for storage within the filesystem of the map-reduce framework can be executed efficiently, many computations do not match this framework and/or are not easily adapted to have all their inputs and outputs within the distributed filesystem.

In a general aspect, a method for managing a distributed data processing system including a number of processing nodes includes maintaining a number of data stores in the system, each data store of the number of data stores being associated with a corresponding processing node of the number of processing nodes and being associated with a durability level of a number of durability levels, the number of durability levels including a first durability level and a second durability level with a relatively greater degree of durability than the first durability level. The method also includes processing a number of sets of data units using two or more processing nodes of the number of processing nodes, each data unit of each set of data units being associated with a corresponding time interval of a number of time intervals. The number of sets of data units includes a first set of data units associated with a first time interval of the number of time intervals.

The processing includes, for each particular durability level, updating an associated indicator to indicate that all sets of data units associated with the first time interval are stored at that particular durability level. The processing also includes processing a number of sets of requests using two or more of the number of processing nodes, each request of each set of requests being configured to cause a state update at a processing node of the number of processing nodes and being associated with a corresponding time interval of the number of time intervals, the number of sets of requests including a first set of requests associated with a second time interval of the number of time intervals. The processing also includes maintaining at a first processing node of the number of processing nodes, a number of counters.

The number of counters includes a working counter indicating a current time interval of the number of time intervals in the distributed data processing system, and a replication counter indicating a time interval of the number of time intervals for which all requests associated with that time interval are replicated at multiple processing nodes of the number of processing nodes.

The method also includes providing, a first message from the first processing node to the other processing nodes of the number of processing nodes at a first time, the first message including the value of the working counter, and the value of the replication counter.

Aspects may include one or more of the following features.

The number of counters may further include a persistence counter indicating a time interval of the number of time intervals for which all requests associated with that time interval are stored in persistent storage associated with at least one processing node of the number of processing nodes. The method may include, for each data unit of the first set of data units, storing the data unit in data stores of the number of data stores associated with respective processing nodes of the number of processing nodes, including storing the data unit in data stores of the number of data stores associated with the first level of durability and storing the data unit in one or more data stores of the number of data stores associated with the second level of durability.

In another general aspect, software for managing a distributed data processing system including a number of processing nodes is stored in a non-transitory form on a computer-readable medium. The software includes instructions for causing a computing system to maintain a number of data stores in the system, each data store of the number of data stores being associated with a corresponding processing node of the number of processing nodes and being associated with a durability level of a number of durability levels, the number of durability levels including a first durability level and a second durability level with a relatively greater degree of durability than the first durability level. The instructions also cause the computing system to process a number of sets of data units using two or more processing nodes of the number of processing nodes, each data unit of each set of data units being associated with a corresponding time interval of a number of time intervals, the number of sets of data units including a first set of data units associated with a first time interval of the number of time intervals, the processing including, for each particular durability level, updating an associated indicator to indicate that all sets of data units associated with the first time interval are stored at that particular durability level.

The instructions also cause the computing system to process a number of sets of requests using two or more of the number of processing nodes, each request of each set of requests being configured to cause a state update at a processing node of the number of processing nodes and being associated with a corresponding time interval of the number of time intervals, the number of sets of requests including a first set of requests associated with a second time interval of the number of time intervals. The instructions also cause the computing system to maintain at a first processing node of the number of processing nodes, a number of counters. The number of counters includes a working counter indicating a current time interval of the number of time intervals in the distributed data processing system, and a replication counter indicating a time interval of the number of time intervals for which all requests associated with that time interval are replicated at multiple processing nodes of the number of processing nodes.

The instructions also cause the computing system to provide a first message from the first processing node to the other processing nodes of the number of processing nodes at a first time, the first message including the value of the working counter, and the value of the replication counter.

In another general aspect, an apparatus includes a distributed data processing system including a number of processing nodes, each processing node including at least one processor and a communication medium connecting the number of processing nodes for sending and receiving information between processing nodes of the number of processing nodes. The distributed data processing system is configured to maintain a number of data stores in the system, each data store of the number of data stores being associated with a corresponding processing node of the number of processing nodes and being associated with a durability level of a number of durability levels, the number of durability levels including a first durability level and a second durability level with a relatively greater degree of durability than the first durability level.

The apparatus is also configured to process a number of sets of data units using two or more processing nodes of the number of processing nodes, each data unit of each set of data units being associated with a corresponding time interval of a number of time intervals, the number of sets of data units including a first set of data units associated with a first time interval of the number of time intervals, the processing including, for each particular durability level, updating an associated indicator to indicate that all sets of data units associated with the first time interval are stored at that particular durability level.

The apparatus is also configured to process a number of sets of requests using two or more of the number of processing nodes, each request of each set of requests being configured to cause a state update at a processing node of the number of processing nodes and being associated with a corresponding time interval of the number of time intervals, the number of sets of requests including a first set of requests associated with a second time interval of the number of time intervals. The apparatus is also configured to maintain at a first processing node of the number of processing nodes, a number of counters.

The number of counters includes a working counter indicating a current time interval of the number of time intervals in the distributed data processing system, and a replication counter indicating a time interval of the number of time intervals for which all requests associated with that time interval are replicated at multiple processing nodes of the number of processing nodes. The apparatus is also configured to provide, a first message from the first processing node to the other processing nodes of the number of processing nodes at a first time, the first message including the value of the working counter, and the value of the replication counter.

In another general aspect, a method for managing a distributed data processing system including a number of processing nodes includes receiving input data at a distributed data processing system interface component in communication with the distributed data processing system, providing the received input data to the distributed data processing system, wherein the distributed data processing system assigns an indicator associated with a first time interval of a number of time intervals to the input data, receiving, at the distributed data processing system interface component, result data associated with the input data from the distributed data processing system, wherein the result data includes the indicator associated with the first time interval, determining, at a first processing node of the number of processing nodes, an indicator associated with a second time interval of the number of time intervals, comparing, at the distributed data processing system interface component, the indicator associated with the second time interval to the indicator associated with the first time interval included in the result data and, if the indicator associated with the second time interval corresponds to a time interval identical to or later than the first time interval, releasing the result data from the distributed data processing system interface component, maintaining a number of data stores in the system, each data store of the number of data stores being associated with a corresponding processing node of the number of processing nodes and being associated with a durability level of a number of durability levels, the number of durability levels including a first durability level and a second durability level with a relatively greater degree of durability than the first durability level, and processing a number of sets of data units using two or more processing nodes of the number of processing nodes, each data unit of each set of data units being associated with a corresponding time interval of the number of time intervals, the number of sets of data units including a first set of data units associated with a third time interval of the number of time intervals, the processing including, for each particular durability level, updating an associated indicator to indicate that all sets of data units associated with the third time interval are stored at that particular durability level.

Aspects may include one or more of the following features.

For each data unit of the first set of data units, the data unit may be stored in data stores of the number of data stores associated with respective processing nodes of the number of processing nodes, including storing the data unit in data stores of the number of data stores associated with the first level of durability and storing the data unit in one or more data stores of the number of data stores associated with the second level of durability. The indicator associated with the second time interval may be provided to the distributed data processing system interface component.

In another general aspect, software for managing a distributed data processing system including a number of processing nodes is stored in a non-transitory form on a computer-readable medium. The software includes instructions for causing a computing system to receive input data at a distributed data processing system interface component in communication with the distributed data processing system, provide the received input data to the distributed data processing system, wherein the distributed data processing system assigns an indicator associated with a first time interval of a number of time intervals to the input data, receive, at the distributed data processing system interface component, result data associated with the input data from the distributed data processing system, wherein the result data includes the indicator associated with the first time interval, determine, at a first processing node of the number of processing nodes, an indicator associated with a second interval of the number of time intervals, compare, at the distributed data processing system interface component, the indicator associated with the second time interval to the indicator associated with the first time interval included in the result data and, if the indicator associated with the second time interval corresponds to a time interval identical to or later than the first time interval, releasing the result data from the distributed data processing system interface component, maintain a number of data stores in the system, each data store of the number of data stores being associated with a corresponding processing node of the number of processing nodes and being associated with a durability level of a number of durability levels, the number of durability levels including a first durability level and a second durability level with a relatively greater degree of durability than the first durability level, and process a number of sets of data units using two or more processing nodes of the number of processing nodes, each data unit of each set of data units being associated with a corresponding time interval of the number of time intervals, the number of sets of data units including a first set of data units associated with a third time interval of the number of time intervals, the processing including, for each particular durability level, updating an associated indicator to indicate that all sets of data units associated with the third time interval are stored at that particular durability level.

In another general aspect, an apparatus includes a distributed data processing system including a number of processing nodes, each processing node including at least one processor, and a communication medium connecting the number of processing nodes for sending and receiving information between processing nodes of the number of processing nodes. The distributed data processing system is configured to receive input data at a distributed data processing system interface component in communication with the distributed data processing system, provide the received input data to the distributed data processing system, wherein the distributed data processing system assigns an indicator associated with a first time interval of a number of time intervals to the input data, receive, at the distributed data processing system interface component, result data associated with the input data from the distributed data processing system, wherein the result data includes the indicator associated with the first time interval, determine, at a first processing node of the number of processing nodes, an indicator associated with a second interval of the number of time intervals, compare, at the distributed data processing system interface component, the indicator associated with the second time interval to the indicator associated with the first time interval included in the result data and, if the indicator associated with the second time interval corresponds to a time interval identical to or later than the first time interval, releasing the result data from the distributed data processing system interface component, maintain a number of data stores in the system, each data store of the number of data stores being associated with a corresponding processing node of the number of processing nodes and being associated with a durability level of a number of durability levels, the number of durability levels including a first durability level and a second durability level with a relatively greater degree of durability than the first durability level, and process a number of sets of data units using two or more processing nodes of the number of processing nodes, each data unit of each set of data units being associated with a corresponding time interval of the number of time intervals, the number of sets of data units including a first set of data units associated with a third time interval of the number of time intervals, the processing including, for each particular durability level, updating an associated indicator to indicate that all sets of data units associated with the third time interval are stored at that particular durability level.

In another general aspect, a method for managing a distributed data processing system including a number of processing nodes includes receiving input data at a distributed data processing system interface component in communication with the distributed data processing system, providing the received input data to the distributed data processing system, wherein the distributed data processing system assigns an indicator associated with a first time interval of a number of time intervals to the input data, receiving, at the distributed data processing system interface component, result data associated with the input data from the distributed data processing system, wherein the result data includes the indicator associated with the first time interval, determining, at a first processing node of the number of processing nodes, an indicator associated with a second interval of the number of time intervals, comparing, at the distributed data processing system interface component, the indicator associated with the second time interval to the indicator associated with the first time interval included in the result data and, if the indicator associated with the second time interval corresponds to a time interval identical to or later than the first time interval, releasing the result data from the distributed data processing system interface component, processing a number of sets of requests using two or more of the number of processing nodes, each request of each set of requests being configured to cause a state update at a processing node of the number of processing nodes and being associated with a corresponding time interval of the number of time intervals, the number of sets of requests including a first set of requests associated with a third time interval of the number of time intervals, maintaining at the first processing node of the number of processing nodes, a number of counters, the number of counters. The number of counter includes a working counter indicating a current time interval of the number of time intervals in the distributed data processing system, and a replication counter indicating a time interval of the number of time intervals for which all requests associated with that time interval are replicated at multiple processing nodes of the number of processing nodes. The method also includes providing, a first message from the first processing node to the other processing nodes of the number of processing nodes at a first time, the first message including the value of the working counter, and the value of the replication counter.

Aspects may include one or more of the following features.

The indicator associated with the second time interval may be provided to the distributed data processing system interface component. The number of counters may include a persistence counter indicating a time interval of the number of time intervals for which all requests associated with that time interval are stored in persistent storage associated with at least one processing node of the number of processing nodes.

In another general aspect, software for managing a distributed data processing system including a number of processing nodes is stored in a non-transitory form on a computer-readable medium. The software includes instructions for causing a computing system to receive input data at a distributed data processing system interface component in communication with the distributed data processing system, provide the received input data to the distributed data processing system, wherein the distributed data processing system assigns an indicator associated with a first time interval of a number of time intervals to the input data, receive, at the distributed data processing system interface component, result data associated with the input data from the distributed data processing system, wherein the result data includes the indicator associated with the first time interval, determine, at a first processing node of the number of processing nodes, an indicator associated with a second interval of the number of time intervals, compare, at the distributed data processing system interface component, the indicator associated with the second time interval to the indicator associated with the first time interval included in the result data and, if the indicator associated with the second time interval corresponds to a time interval identical to or later than the first time interval, releasing the result data from the distributed data processing system interface component, process a number of sets of requests using two or more of the number of processing nodes, each request of each set of requests being configured to cause a state update at a processing node of the number of processing nodes and being associated with a corresponding time interval of the number of time intervals, the number of sets of requests including a first set of requests associated with a third time interval of the number of time intervals, maintain at the first processing node of the number of processing nodes, a number of counters. The number of counters includes a working counter indicating a current time interval of the number of time intervals in the distributed data processing system, and a replication counter indicating a time interval of the number of time intervals for which all requests associated with that time interval are replicated at multiple processing nodes of the number of processing nodes. The software also includes instructions for causing the computing system to provide, a first message from the first processing node to the other processing nodes of the number of processing nodes at a first time, the first message including the value of the working counter, and the value of the replication counter.

In another general aspect, an apparatus includes a distributed data processing system including a number of processing nodes, each processing node including at least one processor and a communication medium connecting the number of processing nodes for sending and receiving information between processing nodes of the number of processing nodes. The distributed data processing system is configured to receive input data at a distributed data processing system interface component in communication with the distributed data processing system, provide the received input data to the distributed data processing system, wherein the distributed data processing system assigns an indicator associated with a first time interval of a number of time intervals to the input data, receive, at the distributed data processing system interface component, result data associated with the input data from the distributed data processing system, wherein the result data includes the indicator associated with the first time interval, determine, at a first processing node of the number of processing nodes, an indicator associated with a second interval of the number of time intervals, compare, at the distributed data processing system interface component, the indicator associated with the second time interval to the indicator associated with the first time interval included in the result data and, if the indicator associated with the second time interval corresponds to a time interval identical to or later than the first time interval, releasing the result data from the distributed data processing system interface component, process a number of sets of requests using two or more of the number of processing nodes, each request of each set of requests being configured to cause a state update at a processing node of the number of processing nodes and being associated with a corresponding time interval of the number of time intervals, the number of sets of requests including a first set of requests associated with a third time interval of the number of time intervals, maintain at the first processing node of the number of processing nodes, a number of counters. The number of counter includes a working counter indicating a current time interval of the number of time intervals in the distributed data processing system, and a replication counter indicating a time interval of the number of time intervals for which all requests associated with that time interval are replicated at multiple processing nodes of the number of processing nodes. The apparatus is also configured to provide, a first message from the first processing node to the other processing nodes of the number of processing nodes at a first time, the first message including the value of the working counter, and the value of the replication counter.

In another aspect, in general, managing a distributed data processing system including a number of processing nodes includes: providing input data to the distributed data processing system, wherein the distributed data processing system assigns an indicator associated with a first time interval of a number of time intervals to the input data; receiving result data associated with the input data from the distributed data processing system, wherein the result data includes the indicator associated with the first time interval; determining, at a first processing node of the number of processing nodes, an indicator associated with a second interval of the number of time intervals; comparing the indicator associated with the second time interval to the indicator associated with the first time interval included in the result data and, if the indicator associated with the second time interval corresponds to a time interval identical to or later than the first time interval, releasing the result data; processing a number of sets of data units using two or more processing nodes of the number of processing nodes, each data unit of each set of data units being associated with a corresponding time interval of the number of time intervals, the number of sets of data units including a first set of data units associated with a third time interval of the number of time intervals, the processing including, for each particular durability level of a number of durability levels, updating an associated indicator to indicate that all sets of data units associated with the third time interval are stored at that particular durability level, the number of durability levels including a first durability level and a second durability level with a relatively greater degree of durability than the first durability level; processing a number of sets of requests using two or more of the number of processing nodes, each request of each set of requests being configured to cause a state update at a processing node of the number of processing nodes and being associated with a corresponding time interval of the number of time intervals; maintaining at the first processing node of the number of processing nodes a replication counter indicating a time interval of the number of time intervals for which all requests associated with that time interval are replicated at multiple processing nodes of the number of processing nodes; and providing, a first message from the first processing node to the other processing nodes of the number of processing nodes, the first message including the value of the replication counter.

In another aspect, in general, managing a distributed data processing system including a number of processing nodes includes: providing input data to the distributed data processing system, wherein the distributed data processing system assigns an indicator associated with a first time interval of a number of time intervals to the input data; receiving result data associated with the input data from the distributed data processing system, wherein the result data includes the indicator associated with the first time interval; determining, at a first processing node of the number of processing nodes, an indicator associated with a second interval of the number of time intervals; releasing the result data based on comparing the indicator associated with the second time interval to the indicator associated with the first time interval included in the result data; processing a number of sets of data units using two or more processing nodes of the number of processing nodes, each data unit of each set of data units being associated with a corresponding time interval of the number of time intervals, the number of sets of data units including a first set of data units associated with a third time interval of the number of time intervals, the processing including, for each particular durability level of a number of durability levels, updating an associated indicator to indicate that all sets of data units associated with the third time interval are stored at that particular durability level. Processing a number of sets of requests using two or more of the number of processing nodes, each request of each set of requests being configured to cause a state update at a processing node of the number of processing nodes and being associated with a corresponding time interval of the number of time intervals, wherein state updates at one or more of the number of processing nodes include updating state stored in memory that is managed using memory management that avoids automatic garbage collection (e.g., explicit memory allocation and deallocation or automatic reference counting); maintaining at the first processing node of the number of processing nodes a replication counter indicating a time interval of the number of time intervals for which all requests associated with that time interval are replicated at multiple processing nodes of the number of processing nodes; and providing, a first message from the first processing node to at least one other processing node of the number of processing nodes, the first message including the value of the replication counter.

Aspects can have one or more of the following advantages.

In general, some features described herein enable an increase computational efficiency (e.g., a distributed data processing system that includes a number of processing nodes is able to increase a number of records processed per unit of given computing resources) of a computation, especially a computation whose underlying specification is in terms of a graph-based program specification, as compared to approaches described above, in which components (or parallel executing copies of components) are hosted on different servers. For example, a call cluster component is disposed in a graph-based program specification and is used to interface the graph-based program specification with the distributed data processing system such that computations required by the graph-based program specification are performed in a distributed manner by the processing nodes in the graph-based program specification. Furthermore, some features described herein provide the ability to adapt to varying computation resources and computation requirements. A computation approach is provided herein that permits adapting to variation in the computing resources that are available during execution of one or more graph-based computations, and/or to variations in the computation load or time variation of load of different components of such computations, for example, due to characteristics of the data being processed. For example, aspects are able to adapt to processing nodes being added or removed (or failing and coming back online) from the distributed data processing system. One way that the distributed data processing system provides the adaptation is by managing replication and persistence of data in the system including maintaining counts of messages sent and received by processing nodes and maintaining indicators of time intervals where all messages are replicated and/or made persistent in the system.

A computation approach is also provided that is able to efficiently make use of computational resources with different characteristics, for example, using servers that have different numbers of processors per server, different numbers of processor cores per processor, etc., and to support both homogeneous as well as heterogeneous environments efficiently. Some features described herein are also able to make the start-up of graph-based computations quick. One aspect of providing such efficiency and adaptability is providing for appropriate management of a cluster of processing nodes, as described herein.

Aspects also are advantageously fault tolerant in that the distributed data processing system is able to recover from any processing errors that occur by rolling the processing back in time. The system anticipates a number of possible rollback scenarios and implements algorithms for performing the rollback in each of the possible rollback scenarios.

