HIERARCHICAL ORGANIZATION FOR SCALE-OUT CLUSTER

• US 20150188978A1
• Filed: 12/30/2013
• Published: 07/02/2015
• Est. Priority Date: 12/30/2013
• Status: Active Grant

• Current Assignee: Microsoft Technology Licensing LLC (Microsoft Corporation)
• Original Assignee: Microsoft Corporation
• Inventors: Petculescu, Cristian, Netz, Amir
• Granted Patent: US 9,723,054 B2
• Time in Patent Office: Days
• US Class Current: 1/1
• 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. In a distributed computing environment, a method of performing a distributed data operation, the method comprising:

receiving a request for one or more data operations from a first computing system;

determining a number of node endpoints that can be used to satisfy the query;

based on the number of node endpoints, 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; and

servicing the request for one or more data operations using the selected data operation plan.

2. The method of claim 1, further comprising determining the node endpoints hosting shards needed to satisfy the data operations, and wherein determining a number of node endpoints is based on the node endpoints on which the shards are hosted.

3. The method of claim 2, further comprising determining that the plan should be at least partially hierarchical, and selecting a number of intermediary node endpoints.

4. The method of claim 3, wherein the intermediary node endpoints are selected based on the location of node endpoints hosting shards.

5. The method of claim 3, wherein the intermediary node endpoints are selected based on a cost determination including factors including communication costs to hierarchical nodes and cost to assemble distributed query results.

6. The method of claim 5, further comprising selecting intermediary node endpoints in a round robin fashion.

7. The method of claim 5, further comprising selecting intermediary node endpoints in a random fashion.

8. The method of claim 3, wherein the plan is configured such that a substantially even distribution of node endpoints is made across intermediary node endpoints.

9. In a distributed computing environment, a system for performing a distributed data operation, the system comprising:

one or more processors; and

one or more computer readable media, wherein the one or more computer readable media comprise computer executable instructions that when executed by at least one of the one or more processors cause at least one of the one or more processors to perform the following;

receiving a request for one or more data operations from a first computing system;

determining a number of node endpoints that can be used to satisfy the query;

based on the number of node endpoints, 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; and

servicing the request for one or more data operations using the selected data operation plan.

10. The system of claim 9, wherein the one or more computer readable media comprise computer executable instructions that when executed by at least one of the one or more processors cause at least one of the one or more processors to determine the node endpoints hosting shards needed to satisfy the data operations, and wherein determining a number of node endpoints is based on the node endpoints on which the shards are hosted.

11. The system of claim 10, wherein the one or more computer readable media comprise computer executable instructions that when executed by at least one of the one or more processors cause at least one of the one or more processors to determine that the plan should be at least partially hierarchical, and select a number of intermediary node endpoints.

12. The system of claim 11, wherein the intermediary node endpoints are selected based on the location of node endpoints hosting shards.

13. The system of claim 11, wherein the intermediary node endpoints are selected based on a cost determination including factors including communication costs to hierarchical nodes and cost to assemble distributed query results.

14. The system of claim 13, wherein the one or more computer readable media comprise computer executable instructions that when executed by at least one of the one or more processors cause at least one of the one or more processors to select intermediary node endpoints in a round robin fashion.

15. The system of claim 13, wherein the one or more computer readable media comprise computer executable instructions that when executed by at least one of the one or more processors cause at least one of the one or more processors to select intermediary node endpoints in a random fashion.

16. The system of claim 11, wherein the plan is configured such that a substantially even distribution of node endpoints is made across intermediary node endpoints.

17. A physical computer readable storage medium comprising computer executable instructions that when executed by one or more processors cause the following to be performed:

receiving a request for one or more data operations from a first computing system;

determining a number of node endpoints that can be used to satisfy the query;

based on the number of node endpoints, 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; and

servicing the request for one or more data operations using the selected data operation plan.

18. The computer readable storage medium of claim 17, further comprising computer executable instructions that when executed by one or more processors cause determining the node endpoints hosting shards needed to satisfy the data operations, and wherein determining a number of node endpoints is based on the node endpoints on which the shards are hosted.

19. The computer readable storage medium of claim 18, further comprising computer executable instructions that when executed by one or more processors cause determining that the plan should be at least partially hierarchical, and selecting a number of intermediary node endpoints.

20. The computer readable storage medium of claim 19, wherein the intermediary node endpoints are selected based on the location of node endpoints hosting shards.