Hierarchical organization for scale-out cluster

• US 9,723,054 B2
• Filed: 12/30/2013
• Issued: 08/01/2017
• Est. Priority Date: 12/30/2013
• Status: Active Grant

• Current Assignee: Microsoft Technology Licensing LLC (Microsoft Corporation)
• Original Assignee: Microsoft Technology Licensing LLC (Microsoft Corporation)
• Inventors: Petculescu, Cristian, Netz, Amir
• Primary Examiner(s): Dollinger, Tonia L
• Assistant Examiner(s): Dabipi, Dixon
• Application Number:
  US14/144,355
• Publication Number:
  US 20150188978A1
• Time in Patent Office: 1,310 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 one or more data operations, the one or more data operations being performed by at least some of a plurality of node endpoints, wherein each node endpoint of the plurality includes a partition of a sharded table;

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

after determining how many node endpoints are needed to satisfy the request, servicing the request to perform the one or more data operations, wherein servicing the request to perform the one or more data operations includes;

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

from within the particular set, designate 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

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

2. The method of claim 1, wherein the method further includes selecting a partially flat/partially hierarchical data operation plan to service the request from a plurality of operation plans, and wherein a number of intermediary node endpoints are selected to implement the partially hierarchical data operation plan.

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

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

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

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

7. The method of claim 2, 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, and wherein the selected data operation plan is configured such that a substantially even distribution of node endpoints is made across intermediary node endpoints.

8. 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 distributed data operation by causing the computer system to;

receive a request to perform one or more data operations, the one or more data operations being performed by at least some of a plurality of node endpoints, wherein each node endpoint of the plurality includes a partition of a sharded table;

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

after determining how many node endpoints are needed to satisfy the request, service the request to perform the one or more data operations, wherein servicing the request to perform the one or more data operations includes;

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

from within the particular set, designate 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

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

9. The computer system of claim 8, wherein the method further includes selecting a partially hierarchical data operation plan to service the request from a plurality of operation plans, and wherein a number of intermediary node endpoints are selected to implement the partially hierarchical data operation plan.

10. The computer system of claim 9, wherein the intermediary node endpoints are selected based on a location of node endpoints hosting shards.

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

12. The computer system of claim 11, wherein the computer executable instructions further cause the computer system to select intermediary node endpoints in a round robin fashion.

13. The computer system of claim 11, wherein the computer executable instructions further cause the computer system to select intermediary node endpoints in a random fashion.

14. The computer 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, and wherein the selected data operation plan is configured such that a substantially even distribution of node endpoints is made across intermediary node endpoints.

18. The computer system of claim 8, wherein the one or more data operations include 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 8, wherein the one or more data operations include a query operation, and wherein servicing the query operation includes:

causing the master node endpoint to request at least some 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.

15. 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 one or more data operations, the one or more data operations being performed by at least some of a plurality of node endpoints, wherein each node endpoint of the plurality includes a partition of a sharded table;

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

after determining how many node endpoints are needed to satisfy the request, service the request to perform the one or more data operations, wherein servicing the request to perform the one or more data operations includes;

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

from within the particular set, designate 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

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

16. The one or more hardware storage device of claim 15, wherein the method further includes selecting a partially flat/partially hierarchical data operation plan to service the request from a plurality of operation plans, and wherein a number of intermediary node endpoints are selected to implement the partially hierarchical data operation plan.

17. The one or more hardware storage device in accordance with claim 16, wherein the intermediary node endpoints are selected based on a location of node endpoints hosting shards.