DATA PLACEMENT CONTROL FOR DISTRIBUTED COMPUTING ENVIRONMENT

• US 20170031988A1
• Filed: 07/30/2015
• Published: 02/02/2017
• Est. Priority Date: 07/30/2015
• Status: Active Grant

• Current Assignee: Futurewei Technologies Incorporated (Huawei Investment & Holding Co., Ltd.)
• Original Assignee: Futurewei Technologies Incorporated (Huawei Investment & Holding Co., Ltd.)
• Inventors: Sun, Jason Yang, Zhang, Guogen, Cai, Le
• Granted Patent: US 10,055,458 B2
• Time in Patent Office: Days
• US Class Current: 1/1
• CPC Class Codes: G06F 16/24544 Join order optimisation
• CPC Class Codes: G06F 16/24554 Unary operations; Data partitioning operations

Abstract

A method includes dividing a dataset into partitions by hashing a specified key, selecting a set of distributed file system nodes as a primary node group for storage of the partitions, and causing a primary copy of the partitions to be stored on the primary node group by a distributed storage system file server such that the location of each partition is known by hashing of the specified key.

Claims

1. A method comprising:

dividing a dataset into partitions by hashing a specified key;

selecting a set of distributed storage system nodes as a primary node group for storage of the partitions; and

causing a primary copy of the partitions to be stored on the primary node group by a distributed storage system file server such that the location of each partition is known by hashing of the specified key.

2. The method of claim 1 and further comprising:

dividing at least one additional dataset into partitions by hashing the specified key, wherein the datasets comprise tables; and

causing a primary copy of the partitions of each additional dataset to be stored on corresponding primary node groups by a distributed storage system file server as a function of the hash such that the location of each partition is known.

3. The method of claim 2 and further comprising optimizing a query utilizing the known locations of partitions.

4. The method of claim 2 and further comprising optimizing a query that includes a join over the tables by performing data shuffles between the partitions where partition groups are different or contained striped partitions.

5. The method of claim 2 wherein the number of partitions for each of the tables is a power of two.

6. The method of claim 5 wherein at least one partition is striped across multiple nodes of the primary node group.

7. The method of claim 2 and further comprising causing a replica copy of the partitions to be stored on a replica node group by the distributed storage system file server wherein each node in the replica node group is not one to one mapped with nodes in the primary node group.

8. The method of claim 2 and further comprising causing a replica copy of the partitions to be stored in accordance with heuristics of the distributed storage system file server.

9. The method of claim 2 wherein the distributed storage system comprises a Hadoop system and wherein causing a primary copy of the partitions to be stored on the primary node group comprises identifying nodes in a partition group associated with a table to a Hadoop file server using a pluggable BlockPlacementPolicy.

10. A system comprising:

a processor; and

a memory device coupled to the processor and having code stored therein to cause the processor to perform a method, the method comprising;

dividing each of multiple datasets into partitions by hashing a specified key;

selecting sets of distributed storage system nodes as primary node groups for storage of the partitions; and

causing a primary copy of the partitions of each dataset to be stored on corresponding primary node groups by a distributed storage system file server such that the location of each partition is known by hashing of the specified key.

11. The system of claim 10 wherein the memory device further comprises code to cause the processor to perform as a query coordinator to communicate with the primary node groups and optimize queries based on the known locations of each partition.

12. The system of claim 11 wherein optimizing a query that includes a join over the multiple datasets is performed by specifying data shuffles between the partitions where partition groups are different or contained striped partitions.

13. The system of claim 10 wherein the number of partitions for each of the multiple datasets is a power of two.

14. The system of claim 13 wherein at least one partition is striped across multiple nodes.

15. The system of claim 10 wherein the method further comprises causing a replica copy of the partitions to be stored on a replica node group by the distributed storage system file server wherein each node in the replica node group is not one to one mapped with nodes in the primary node group.

16. The system of claim 10 wherein the method further comprises causing a replica copy of the partitions to be stored in accordance with heuristics of the distributed storage system file server.

17. The system of claim 10 wherein the distributed storage system comprises a Hadoop system and wherein causing a primary copy of the partitions to be stored on the primary node group comprises identifying nodes in a partition group associated with a file to a Hadoop file server using a pluggable BlockPlacementPolicy.

18. A method comprising:

receiving a query;

inspecting the query to identify tables being joined and join conditions;

identifying partition groups organized using distributed hash tables on a partition key and their sizes associated with multiple tables corresponding to the join conditions;

performing a collocated join plan if an equi-join predicate is on partition keys;

constructing a collocated join plan if the partition groups are the same;

constructing a partition-wise join plan for bracketed exchange between nodes if the partition groups are the same; and

if the size of the partition groups is not the same;

constructing one or more logic partitions to match partitions in the groups;

mapping partitions in a larger group to partitions in a smaller group; and

constructing a partition-wise join plan for bracketed exchange between the nodes in the large and small groups based on the maps.

19. The method of claim 18 wherein the number of partitions within a group are a power of two and wherein the partition key is a primary key.

20. The method of claim 18 wherein K is defined as the number of partitions in the larger group divided by the number of partitions in the smaller group, and wherein each partition in the smaller group is mapped to K partitions of the larger group.