Distributed partitioned map reduce using a data fabric

A system for a distributed partitioned map reduce includes a plurality of nodes, a plurality of partitions, and a service node. The one or more partitions of the plurality of partitions are stored on a first node of the plurality of nodes. The service node is configured to receive a task from a client application system. The client application system receives a request to perform the task from a requestor. The client application system selects a service node to manage execution of the task. The client application system provides the service node the task. The service node is configured to provide partition task logic to the plurality of partitions. The partition task logic includes a partition task map reduce. The service node is configured to receive partition results from the plurality of partitions, execute a service node reduce to determine a service node result, and provide the service node result.

BACKGROUND OF THE INVENTION

Cluster computing systems comprising a plurality of computing systems operating in parallel often use a MapReduce programming paradigm for processing large data sets. The MapReduce paradigm, or more generally map reduce as referred to herein, comprises a map step that is executed in parallel on subsets of a data set (e.g., equal subsets of the data), followed by a reduce step that combines the results of the map step into a single output. The reduce step is executed by the master node of the cluster system. The map step can take full advantage of the parallelism of the cluster, however, the reduce step is executed by the single master node. This creates a limitation where the master node acts as a bottleneck to future cluster computing operations.

DETAILED DESCRIPTION

A system for a distributed partitioned map reduce is disclosed. The system comprises a plurality of nodes, a plurality of partitions, and a service node. One or more partitions of the plurality of partitions are stored on a first node of the plurality of nodes. The service node is configured to receive a task from a client application system, wherein the client application system receives a request to perform the task from a requestor, wherein the client application system selects a service node to manage execution of the task, and wherein the client application system provides the service node the task. The service node is configured to provide partition task logic to the plurality of partitions, wherein the partition task logic includes a partition task map reduce; receive partition results from the plurality of partitions; execute a service node reduce to determine a service node result; and provide the service node result.

A system for a distributed partitioned map reduce is disclosed. The system comprises a plurality of nodes, a plurality of partitions, and one or more services (hosted by one or more service nodes). The partitions are equally divided between the nodes of the system. The service nodes expose a well-defined interface to the system with which they can receive and process requests from a requestor. Services are computational libraries that provide a well-defined interface to the system through which the services can receive requests. Nodes contain services and the same service can be deployed to multiple nodes allowing for scalability. The requestor selects a service node to manage execution of a task and sends the request to that node by using a local representation of the interface exposed by the service running on the service node. When selecting a service node, the system can be configured to use the same node for every request or a different node for every request (e.g., to evenly distribute the load). If a service is deployed to multiple nodes, the selection of a service node is picked at random to start, subsequent requests are routed either to the same node for every request or a different node for every request. The service nodes are configured to receive a task, provide partition task logic to the plurality of partitions, wherein the partition task logic includes a partition task map and a partition task reduce, receive partition results from the plurality of partitions, receive partition results from the plurality of partitions, execute a service node reduce to determine a service node result, and provide the service node result. In some cases, the service node receives a request from the requestor, divides the request equally between the plurality of partitions, and sends the resulting sub-divided tasks to the plurality of nodes. Each node responds to the requestor with each of the results of the requests to which they were assigned, after which the service node aggregates the partitioned results into a single result and responds to the requestor with that result.

A system for a distributed partitioned map reduce comprises a cluster system comprising a plurality of computing nodes and one or more client application systems. A user system communicates with the cluster system via a client application system to request execution of a cluster computing task. The client application system chooses a computing node of the plurality of computing nodes and designates the chosen computing node as the service node for the task. The client application system chooses the computing node in such a way as to balance the load on the nodes of the cluster system. For example, the client application system chooses the least loaded node, the node that least recently acted as a service node, etc. The client application system then provides the designated service node with the task. The task comprises a partition map reduce step which is subdivided by the service node, one per partition, and a final service node reduce step.

The service node receives the task and provides each other node of the cluster system with partition task logic, comprising a partition task map and a partition task reduce. For example, the service node provides the partition nodes (e.g., nodes in the cluster that store partitions, where the partitions collectively are used for the storage of data) in the cluster with the map reduce steps in parallel (one per partition) and then aggregates the results into a single result which it responds with to the requestor. Note that service nodes can also store or not store data in partitions. A partition node receives the partition task logic and executes the partition task map step on each partition stored by the node. It should be noted that the ownership of partitions by nodes is flexible and can be rearranged by the system; a given partition may, at some time, be moved to be stored on a different node, which then assumes ownership of the moved partition. The partition task map step comprises data processing instructions that can be executed on each data element stored by a partition. The partition task reduce is then used to combine the results of executing the partition task map step on elements of data stored in the partition. A partition task reduce is used to generate a partition result associated with each partition stored on the node. The partition results are then provided by the nodes to the service node.

In some embodiments, the partition results are combined using a node reduce step. The node reduce step combines results of each partition stored on a node, generating a single node result associated with each node of the cluster. The node results are then provided by the nodes to the service node.

The service node receives results (e.g., partition results, node results) and combines the results using a service node reduce step to determine a service node result. The service node result comprises the result of the requested parallel clustered map reduce steps. The service node provides the service node result to the requesting system. The system for a distributed partitioned map reduce improves the computing system by distributing the work of the reduce step in two ways, avoiding the bottleneck effect. The system for a distributed partitioned map reduce distributes the reduce step in a first way by selecting the service node to balance the load of the reduce step. A first service node for a first task and a second service node for a second task comprise different nodes, avoiding the bottleneck. Additionally, the partition task reduce step or the node reduce step are distributed across the nodes of the cluster, leaving a minimum of work remaining for the service node to perform.

FIG.1is a block diagram illustrating an embodiment of a network system. In some embodiments, the network system ofFIG.1comprises a system for a distributed partitioned map reduce using a data fabric. Cluster computing system104comprises a computing system for processing cluster computing jobs. Processing cluster computing jobs comprises processing software jobs utilizing a cluster computing system (e.g., a computing system utilizing a plurality of computers operating in parallel). Cluster computing system104comprises a node pool comprising a plurality of computing nodes, wherein each node comprises data storage and data processing capacity. The computing nodes process computing tasks (e.g., reading data, writing data, processing data, etc.). Each node stores one or more partitions, each partition comprising partition data. A plurality of partitions is stored by cluster system104. The cluster computing system additionally comprises one or more client nodes that include a client application system that a user interacts with from user system106. User system106, administrator system102, and cluster system104interact using network100.

Cluster system104comprises a system for a distributed partitioned map reduce using a data fabric. When a cluster computing task is received by a client application system of cluster system104(e.g., from user system106, administrator system102, etc.), the client application system determines a node of a node pool and designates a service node for the task. The service node comprises a node for coordinating execution of the task and for performing a service node reduce step to determine a service node result. The client application system chooses the service node in such a way as to balance the load on the nodes. For example, the client application system chooses the least loaded node, the client application system chooses the node that was least recently chosen as a service node, the client application system chooses the service node according to a round robin ordering, etc. The client application system indicates to that node that it has been designated the service node for the task and provides the task to the service node. The service node, upon receiving the task from the client application system, begins execution and coordination of the task. The service node provides partition task logic to the plurality of partitions stored by cluster system104(e.g., by providing the partition task logic to the plurality of nodes each with their set of partitions). The partition task logic is executed for each partition. The partition task logic comprises a map step associated with a partition that is executed on each data element stored by the partition to create a set of values for each partition. The partition task logic additionally comprises a reduce step that is used to combine values from the map step associated with the partition to create a partition result for each partition stored locally on the node. Some systems for a distributed partitioned map reduce using a data fabric additionally combine the partition results for each partition of a node using a node reduce to generate a node result. The partition results or the node results are provided by the nodes to the service node. The service node then executes the service node reduce to combine the results received from the nodes and determine a service node result. The service node result comprises the task result. The service node provides the service node result to the client application system, which then provides the service node result to the requesting system.

User system106comprises a system for requesting computations to be performed using cluster system104. In various embodiments, network100provides a means for communicating between administrator system102, cluster system104, and user system106, and comprises one or more of the following: a local area network, a wide area network, a wired network, a wireless network, the Internet, an intranet, a storage area network, or any other appropriate communication network. Administrator system102enables an administrator to maintain cluster system104. Administrator system102comprises a system for executing administrator commands, for configuring cluster system104, for querying cluster system104, etc. In some embodiments, user system106comprises a client system, for example an object transaction server or an object read server.

FIG.2is a block diagram illustrating an embodiment of a cluster system. In some embodiments, cluster system200comprises cluster system104ofFIG.1. In some embodiments, cluster system200comprises a data fabric. In the example shown, cluster system200comprises communication bus202for allowing a plurality of nodes to communicate with each other using a network. Nodes of node pool204include client nodes (e.g., client node206and client node216), service nodes (e.g., service node214), and partition nodes (e.g., partition node208, partition node210, partition node212, etc.). Cluster system200comprises any appropriate number of client nodes (e.g., 1 client node, 2 client node, 5 client nodes, etc.). In some embodiments, client nodes of cluster system200are able to interact with any appropriate service nodes in order to balance the task processing load. Communication bus202also enables communication of cluster system200with a network enabling communication with an administrator system, user system, or other cluster system. Each of the one or more nodes comprises a computing system, for example, a computing system comprising a processor, a memory, data storage, etc. Nodes of the node pool are able to communicate with each other via communication bus202. Nodes of the node pool can be separate computer systems, physically remote separate computer systems, separate computer systems connected via the Internet, cloud computing systems, virtual machines, virtual machines running on a single computer, etc. Node pool204comprises any appropriate number of nodes (e.g., 1 node, 2 nodes, 6 nodes, 12 nodes, 23 nodes, etc.). Each node stores data in a plurality of partitions. Some partitions comprise primary partitions. Some partitions comprise backup partitions—for example, copies of primary partitions, wherein a backup partition is stored on a different node than its associated primary partition. A node storing a primary partition and a node storing the associated backup partition can be physically proximate, physically remote, running as virtual machines on the same computer, or instantiated in any other appropriate way.

When cluster system200receives a request to perform a task, the request is initially received by communication bus202. The requester can comprise an administrator system, a client system, a client server, an object transaction server, an object read server, etc. A client application system selects a service node—for example, a client application system associated with a user system selects a least loaded service node, a least recently chosen node, a randomly chosen node, etc.—and routes the request to the determined service node. The service node provides partition task logic to each partition node of the node pool via communication bus202. Each node (e.g., including the service node) then executes the partition task logic. Executing the partition task logic comprises executing a map step on each partition on the node. In some embodiments, executing the task comprises executing a map step on each primary partition on the node. Executing the map step for a partition comprises executing a map operation on each data element stored by the partition to create a set of map values for the partition. Executing the partition task logic additionally comprises executing a reduce step to combine the map values to create a partition result for a partition. In some embodiments, the partition task logic additionally comprises a node reduce. A node executes the node reduce to combine the partition results for the partitions of a node to create a node result for a node. Results from each node (e.g., a set of partition results or a node result for each node) are provided to the service node via communication bus202. The service node uses a service node reduce to combine the results from each node to determine a service node result. The service node result is provided via communication bus202to a client application system and provided via communication bus202to the user system.

FIG.3is a block diagram illustrating an embodiment of a client node. In some embodiments, client node300comprises a client node ofFIG.2(e.g., client node206or client node216). In the example shown, client node300comprises interface302for interacting with other systems, (e.g., with a network, with a user system via the network, with a node pool, with a node of the node pool via a communication bus, etc.). Interface302is in communication with processor304. Processor304comprises a processor for processing data. Processor304comprises service node selector306for selecting a service node (e.g., from a set of nodes of a node pool). Processor304includes a client application system308(e.g., a read system or a transaction system). Processor304is in communication with data storage310for storing data. In some embodiments, the data fabric with its partitions stores all data for the cluster system so that a local data storage (e.g., data storage310) is not used. In some embodiments, data storage310comprises a temporary storage for data such as a cache or processor memory for use during processing. Service node selector306of processor304receives a request to perform a task from client application system308. The service node selector306selects a service node to process a task (e.g., a least loaded node, a next node in a round robin manner, etc.), and provides the task to the selected service node. After the task is executed as managed by the service node, client node300receives the results and provides them to the user.

FIG.4is a block diagram illustrating an embodiment of a partition node. In some embodiments, partition node400comprises a partition node of node pool204ofFIG.2(e.g., partition node208, partition node210, and partition node212). In the example shown, partition node400comprises interface402for interacting with other systems (e.g., with a client application system). Interface402is in communication with processor404. Processor404comprises a processor for processing data. In the example shown, processor404comprises mapper406for executing a map operation. For example, mapper406comprises a mapper for executing a partition map operation on each data element stored by a partition (e.g., a partition stored by data storage410) to create a set of partition map values for the partition. Processor404additionally comprises reducer408for executing a reduce operation. For example, reducer408executes a partition reduce step to combine partition map values to create a partition result for a partition. In some embodiments, reducer408executes a node reduce step to combine partition results to create a node result for a node. Partition results and/or node results are provided by processor404to a service node via interface402. Processor404is in communication with data storage410. Data storage410comprises communication bus412for allowing partition414, partition416, partition418, or partition map420to communicate with processor404. Partition414, partition416, and partition418comprise partitions for storing partition data. In some embodiments, partition414, partition416, and partition418comprise part of a data fabric. Partition map420comprises a partition map for storing partition metadata. Partition metadata comprises primary partition identifiers, backup partition identifiers, indications of partition lock status (e.g., read locked, write locked), etc. Partition map420acts as a gatekeeper for interactions with data stored on the partitions of a node and, in particular, mediates different requests for accessing the data including reading access, writing access, editing access, or deleting access and whether those requests are allowed or not allowed based on what access is currently occurring for the data.

In the example shown, for processing, partition node400receives partition process instructions via interface402for partitions that are used to execute operations for elements of each partition (e.g., partition414, partition416, and partition418) of partition node400. Mapper406of processor404distributes map instructions to each partition and results are provided to reducer208. The results are reduced at the partition level or further reduced at the node level. The reduced results are provided to the service node. Access to the partitions is coordinated by partition map420. For example, partition map420is used to negotiate access to data stored in a partition. Specifically, one example of access control by partition map420is when reading access is delayed to data in a given partition by partition map420until an existing write for that specific data in the given partition has been completed.

FIG.5is a block diagram illustrating an embodiment of a service node. In some embodiments, service node500comprises a node of node pool204ofFIG.2(e.g., service node214). In the example shown, service node500is selected as a service node by a client application system to perform a task and receives task via interface502. Mapper510of processor504causes execution of a map partition process for each data stored in the partitions (e.g., partition518, partition520, partition522, as well as partitions on other nodes). Reducer512of processor504causes execution of a reduce partition process to create a result. Reducer508of service provider506causes execution of a reduce of results from the partition processes of the partitions or nodes. The result of service provider506reduce is provided via interface502to the client application system, which in turn provides the result to the requestor of the task.

In the example shown, service node500comprises interface502for interacting with other systems (e.g., with a client application system). Interface502is in communication with processor504. Processor504comprises a processor for processing data. Processor504comprises service provider506for coordinating a task. Service provider506comprises reducer508for performing a service node reduce. A service node reduce combines partition results and/or node results to determine a service node result. The service node result is provided by processor504via interface502to a client application system. Processor504additionally comprises mapper510for executing a map operation. For example, mapper510comprises a mapper for executing a partition map operation on each data element stored by a partition (e.g., a partition stored by data storage514) to create a set of partition map values for the partition. Processor504additionally comprises reducer512for executing a reduce operation. For example, reducer512executes a partition reduce step to combine partition map values to create a partition result for a partition. In some embodiments, reducer512executes a node reduce step to combine partition results to create a node result for a node. Partition results and/or node results are provided to service provider506. Processor504is in communication with data storage514. Data storage514comprises communication bus516for allowing partition518, partition520, partition522, or partition map524to communicate with processor504. Partition518, partition520, and partition522comprise partitions for storing partition data. In some embodiments, partition518, partition520, and partition522comprise part of a data fabric. Partition map524comprises a partition map for storing partition metadata. Partition metadata comprises primary partition identifiers, backup partition identifiers, indications of partition lock status (e.g., read locked, write locked), etc. Partition map524acts as a gatekeeper for interactions with data stored on the partitions of a node and, in particular, mediates different requests for accessing the data including reading access, writing access, editing access, or deleting access and whether those requests are allowed or not allowed based on what access is currently occurring for the data.

FIG.6is a block diagram illustrating an embodiment of a partition. In some embodiments, partition600comprises a partition ofFIG.4or a partition ofFIG.5. In the example shown, partition600stores data in partition data device602. Data is processed using partition processing instructions (e.g., a map instruction or a reduce instruction) and the results are provided using the communications bus. Partition data device602comprises a segment of stored cluster data (e.g., data stored on a cluster system—for example, cluster system104ofFIG.1). In various embodiments, partition data comprises raw data, table rows, key-value pairs, or any other appropriate data type. In some embodiments, partition data comprises a set of related data instances—for example, data describing a set of bank accounts, data describing a set of user preferences, data describing a set of transactions, data describing a set of user actions, etc.

FIG.7is a block diagram illustrating an embodiment of a partition map. In some embodiments, partition map700comprises partition map420ofFIG.4or partition map524ofFIG.5. In the example shown, a node receives partition process instructions and causes execution of the instructions to process data in a partition. The partition is processed after locking the data appropriately (e.g., read lock, write lock, etc.) and then after processing unlocking the partition. Partition map700comprises a set of partition identifiers and associated partition lock statuses (e.g., partition identifier702and associated partition lock status704, partition identifier706and associated partition lock status708, and partition identifier710and associated partition lock status712). Each partition identifier stored by partition map700is associated with a partition stored by the same node as partition map700. Partition map700is used to identify the partitions that are stored by the node. Partition map700is additionally used to store a lock status associated with each partition of the node as well as any additional computed values about the contents of the partition data. The lock status comprises a status indicating whether the partition is locked, for example, read locked or write locked. In some embodiments, a partition is locked during a partition map operation.

FIG.8is a flow diagram illustrating an embodiment of a process for a distributed partitioned map reduce using a data fabric. In some embodiments, the process ofFIG.8is executed by a service node ofFIG.2(e.g., service node214). In the example shown, in800, a task is received from a client application system, wherein the client application system receives a request to perform the task from a requestor, wherein the client application system selects a service node to manage execution of the task, and wherein the client application system provides the service node the task. In802, partition task logic is provided to a plurality of partitions, wherein the partition task logic includes a partition task map reduce. For example, the partition logic is distributed to the partitions and the partition logic includes a map step and a reduce step that is to be executed on data stored or associated with the partition. In some embodiments, the plurality of partitions comprises partitions stored on nodes other than the service node (e.g., partition nodes). In some embodiments, the plurality of partitions comprises partitions stored on the service node. In804, partition results are received from the plurality of partitions. In806, a service node reduce is executed to determine a service node result. In808, the service node result is provided (e.g., to the client application system).

FIG.9is a flow diagram illustrating an embodiment of a process for a service node selector. In some embodiments, the process ofFIG.9is executed by a service node selector ofFIG.3(e.g., service node selector306). In the example shown, in900, a request to perform a task is received from a requester. For example, a request is received from an application. In902, a service node is selected to manage execution of the task. In904, the service node is provided the task. In906, the service node result is received. In908, the service node result is provided to the requester. For example, a result is provided to the requesting application (e.g., a client application system).

FIG.10is a flow diagram illustrating an embodiment of a process for executing partition task logic. In some embodiments, the process ofFIG.10is executed by a partition node ofFIG.2(e.g., partition node208, partition node210, or partition node212). In1000, partition task logic is received from a service node. For example, a partition node receives partition task logic from a service node. In1002, it is determined whether the partition task logic indicates to lock the partition. In the event it is determined that the partition task logic indicates to lock the partition, control passes to1004. In1004, the partition is locked using an appropriate lock type—for example, a read lock or a write lock. Locking the partition is indicated to a partition map (e.g., partition map700ofFIG.7). In1006, a partition map step is executed on partition data. For example, a partition node processor or service node processor execute a map step on partition data using a mapper (e.g., mapper406ofFIG.4or mapper510ofFIG.5, respectively). In1008, a partition reduce step is executed on the results of the partition map step. For example, a partition or a service node processor executes a reduce step on partition data using a reducer (e.g., reducer408ofFIG.4or reducer512ofFIG.5, respectively). In1010, the partition is unlocked, and the process ends. In the event that the partition task logic does not indicate to lock the partition, control passes to1012. In1012, a partition map step is executed on partition data. In1014, a partition reduce step is executed on the results of the partition map step.

FIG.11is a flow diagram illustrating an embodiment of a process for a node reduce. In some embodiments, the process ofFIG.8is executed by a node ofFIG.2(e.g., partition node212) for combining partition reduce results to create a node result. In the example shown, in1100, partition results are received from a plurality of partitions stored by the node. In1102, a node reduce is executed on the partition results from the plurality of partitions to determine a node result. In1104, the node result is provided to the service node.