HIERARCHICAL ORGANIZATION FOR SCALE-OUT CLUSTER

• US 20170324801A1
• Filed: 07/25/2017
• Published: 11/09/2017
• Est. Priority Date: 12/30/2013
• Status: Active Application

• Current Assignee: Microsoft Technology Licensing LLC (Microsoft Corporation)
• Original Assignee: Microsoft Technology Licensing LLC (Microsoft Corporation)
• Inventors: Petculescu, Cristian, Netz, Amir
• Granted Patent: US 10,257,255 B2
• Time in Patent Office: Days
• CPC Class Codes: G06F 9/5061 Partitioning or combining of resources
• CPC Class Codes: H04L 67/02 based on web technology, e.g. hypertext transfer protocol [HTTP]
• CPC Class Codes: H04L 67/1089 Hierarchical topologies

Abstract

Performing a distributed data operation. A method includes receiving a request for one or more data operations from a first computing system, such as a client. The method further includes determining a number of node endpoints that can be used to satisfy the query. Based on the number of node endpoints, the method further includes selecting a plan from the group consisting essentially of a flat data operation plan, a hierarchical data operation plan or a combination of partially flat/partially hierarchical data operation plan. The request for one or more data operations is serviced using the selected data operation plan.

Claims

1. A method, implemented at a computer system that includes one or more processors, for performing a distributed data operation, the method comprising the computer system performing the following:

receiving a request to perform an operation, the operation being performed by at least some of a plurality of node endpoints, each node including a partition of a sharded table;

determining how many node endpoints are needed to satisfy the request, wherein the determination is based at least on information obtained from an updateable directory that is consulted in response to the request, the updateable directory mapping the sharded table across the plurality of node endpoints by identifying where each partition of the table is located in each of the plurality of node endpoints;

after determining how many node endpoints are needed to satisfy the request, determining to utilize one or more intermediary nodes alone or in combination with one or more slave nodes to satisfy the request, based at least on a determined computational processing overhead associated with utilizing only slave nodes as compared to utilizing at least one intermediary node to satisfy the request; and

satisfying the request with at least the one or more intermediary nodes.

2. The method of claim 1, wherein the method further includes satisfying the request with the one or more intermediary nodes in combination with the one or more slave nodes.

3. The method of claim 1, wherein the method further includes satisfying the request with the one or more intermediary nodes in combination without the one or more slave nodes.

4. The method of claim 1, wherein the one or more intermediary nodes are selected based on a location of hosted shards.

5. The method of claim 1, wherein the one or more intermediary node endpoints are selected based on a communication costs and a cost to assemble distributed query results.

6. The method of claim 1, further comprising selecting the one or more intermediary nodes in a round robin fashion.

7. The method of claim 1, further comprising selecting the one or more intermediary nodes in a random fashion.

8. The method of claim 1, wherein after determining how many node endpoints are needed to satisfy the request, the method further includes selecting a data operation plan from a group comprising a flat data operation plan, a hierarchical data operation plan, or a combination of partially flat/partially hierarchical data operation plan.

9. In a distributed computing environment, a computer system comprising:

one or more processors; and

one or more computer readable hardware storage devices having stored thereon computer executable instructions that are executable by the one or more processors and that cause the computer system to perform a method comprising the computer system performing the following;

receive a request to perform an operation, the operation being performed by at least some of a plurality of node endpoints, each node including a partition of a sharded table;

determine how many node endpoints are needed to satisfy the request, wherein the determination is based at least on information obtained from an updateable directory that is consulted in response to the request, the updateable directory mapping the sharded table across the plurality of node endpoints by identifying where each partition of the table is located in each of the plurality of node endpoints;

after determine how many node endpoints are needed to satisfy the request, determining to utilize one or more intermediary nodes alone or in combination with one or more slave nodes to satisfy the request, based at least on a determined computational processing overhead associated with utilizing only slave nodes as compared to utilizing at least one intermediary node to satisfy the request; and

satisfy the request with at least the one or more intermediary nodes.

10. The computing system of claim 9, wherein the method further includes satisfying the request with the one or more intermediary nodes in combination with the one or more slave nodes.

11. The computing system of claim 9, wherein the method further includes satisfying the request with the one or more intermediary nodes in combination without the one or more slave nodes.

12. The computing system of claim 9, wherein the one or more intermediary nodes are selected based on a location of hosted shards.

13. The computing system of claim 9, wherein the one or more intermediary node endpoints are selected based on a communication costs and a cost to assemble distributed query results.

14. The computing system of claim 9, further comprising selecting the one or more intermediary nodes in a round robin fashion.

15. The computing system of claim 9, further comprising selecting the one or more intermediary nodes in a random fashion.

16. The computing system of claim 9, wherein after determining how many node endpoints are needed to satisfy the request, the method further includes selecting a data operation plan from a group comprising a flat data operation plan, a hierarchical data operation plan, or a combination of partially flat/partially hierarchical data operation plan.

17. The computer system of claim 9, wherein the request is part of a session of requests, and wherein servicing the request includes:

after determining how many node endpoints are needed to satisfy the request, selecting a particular set of node endpoints that will be used to service the request;

from within the particular set, designating a particular node endpoint to act as a master node endpoint and designate each remaining node endpoint included within the particular set as slave node endpoints; and

directing the request and all subsequent requests included within the session of requests to the master node endpoint.

18. The computer system of claim 17, wherein the operation includes a data processing operation, and wherein servicing the data processing operation includes:

causing a first slave node endpoint to fetch a data item from persistent storage and perform a merge on the data item;

causing a second slave node endpoint to perform a second merge on the data item;

persisting the data item to the persistent storage; and

in response to persisting the data item, causing the master node endpoint to compute a new version of the updateable directory and persisting the new version of the updateable directory to the persistent storage.

19. The computer system of claim 17, wherein the operation includes a query operation, and wherein servicing the query operation includes:

causing the master node endpoint to request at least one of the slave node endpoints to perform one or more sub-queries;

for each slave node endpoint that performed a sub-query, saving results of each sub-query locally on each corresponding slave node endpoint;

in addition to locally saving the results, causing the slave node endpoints to pass the results to the master node endpoint; and

causing the master node endpoint to aggregate the results to form a final result.

20. One or more hardware storage device having stored thereon computer executable instructions that are executable by one or more processors of a computer system to cause the computer system to perform a distributed data operation by at least causing the computer system to:

receive a request to perform an operation, the operation being performed by at least some of a plurality of node endpoints, each node including a partition of a sharded table;

determine how many node endpoints are needed to satisfy the request, wherein the determination is based at least on information obtained from an updateable directory that is consulted in response to the request, the updateable directory mapping the sharded table across the plurality of node endpoints by identifying where each partition of the table is located in each of the plurality of node endpoints;

after determine how many node endpoints are needed to satisfy the request, determining to utilize one or more intermediary nodes alone or in combination with one or more slave nodes to satisfy the request, based at least on a determined computational processing overhead associated with utilizing only slave nodes as compared to utilizing at least one intermediary node to satisfy the request; and

satisfy the request with at least the one or more intermediary nodes.