Data placement control for distributed computing environment

• US 10,055,458 B2
• Filed: 07/30/2015
• Issued: 08/21/2018
• 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
• Primary Examiner(s): Hoang, Son T
• Application Number:
  US14/813,668
• Publication Number:
  US 20170031988A1
• Time in Patent Office: 1,118 Days
• Field of Search: None
• 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 first dataset including a first plurality of elements into partitions by hashing a key for each of the first plurality of elements to generate a hash value for the key of the element, wherein each element of the first plurality of elements is stored in a partition corresponding to the hash value for the key of the element;

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

causing a primary copy of the partitions of the first dataset to be stored on the first primary node group by a distributed storage system file server based on the respective hash values such that the storage system node on which each element of the first plurality of elements is stored is associated with the hash value for the key of the element;

dividing at least one additional dataset into partitions by hashing a key for each element of the at least one additional dataset to generate a hash value for the key of the element, 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 the distributed storage system file server as a function of hash values such that the storage system node of each partition in the corresponding primary node group is known by hashing of the key, wherein a number of partitions that store each of the tables is a power of two, and wherein at least one partition is striped across multiple nodes of the primary node group.

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

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

4. The method of claim 1 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.

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

6. The method of claim 1 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.

7. A system comprising:

a memory having instructions stored thereon; and

a processor in communication with the memory, wherein the processor executes the instructions to;

divide each of multiple datasets including a plurality of elements into partitions by hashing a key for each of the plurality of elements to generate a hash value for the key of the element, wherein each element of the plurality of elements is stored in a partition corresponding to the hash value for the key of the element;

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

cause a primary copy of the partitions of each dataset to be stored on corresponding primary node groups by a distributed storage system file server based on the respective hash values such that the storage system node on which each element of the plurality of elements is stored is associated with the hash value for the key of the element, wherein a number of partitions that store each of the multiple datasets is a power of two, and wherein at least one partition is striped across multiple nodes.

8. The system of claim 7 wherein the processor executes the instructions to perform as a query coordinator to communicate with the primary node groups and optimize queries based on known locations of each partition.

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

10. The system of claim 7 wherein the processor executes the instructions to cause 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.

11. The system of claim 7 wherein the processor executes the instructions to cause a replica copy of the partitions to be stored in accordance with heuristics of the distributed storage system file server.

12. The system of claim 7 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.