Traditionally, large-scale data sets (e.g., on the scale of multiple terabytes (TB) of data) have been stored on disk-based storage systems, such as RAID (Redundant Array of Independent Disks) systems. An alternative to managing large amounts of data using disk-based storage systems is managing the data via distributed in-memory query processing systems (referred to herein as “distributed query processing systems”). A distributed query processing system includes two or more computing devices, where each device stores a portion of the data assigned to the distributed query processing system, and data management is coordinated among all computing devices in the system.
Given the diminishing cost of storage on individual computing devices, it is becoming more cost-effective to use distributed query processing systems to store and process large amounts of data. Also, distributed query processing systems generally are more scalable than disk-based storage systems in that distributed query processing systems comprise a configurable number of storage nodes. Furthermore, the in-memory processing capability of individual computing devices of a distributed query processing system is generally more powerful and efficient than that of a disk-based computing system by an order of magnitude. As such, by using the processing capability of the individual storage nodes in a distributed query processing system for query processing, such a system allows for increased parallelism in processing queries over data stored at the distributed query processing system.
However, distributed query processing systems are subject to some practical constraints. For example, the computing devices in a distributed query processing system are generally constrained in the amount of memory in the devices, i.e., because of the cost of memory (such as, power consumption of the memory), for architectural reasons, etc. Also, in the case of very large data sets, each node in the system stores only a portion of data for a particular data set, and, at times, processing a query over the data set requires replicating data across nodes in the system.
It is important to carefully allocate data, including data for very large data sets, across the computing devices in a distributed query processing system to facilitate management of the data. To that end, data may be distributed among the nodes of a distributed query processing system by observing a placement algorithm. Existing data placement algorithms can be generally classified into two paradigms: centralized data mapping, and decentralized data mapping.
Centralized data mapping assigns a particular node of a cluster to be the “central” node. Centralized data placement schemes are adopted in many systems such as Hadoop Distributed File System (HDFS), General Parallel File System (GPFS), etc.
The assigned central node stores a data map identifying the location of each portion of data stored among the nodes of the system, and also performs data placement decisions. If the assigned central node attempts to place data at a storage node that is not available to store the data, then the central node picks another destination node for the data, e.g., by incrementing through an array of storage node identifiers until the central node identifies an available storage node. Generally, other nodes in the system report to the central node.
However, centralized data mapping involves allocation of significant storage space in the central node to maintain the data map. Further, all of the data placement traffic and the data lookup service traffic condense in the central node, which causes a bottleneck for the whole system. Also, if the central node crashes, then the central data map is destroyed, which compromises the failure resilience of such a system.
Decentralized data mapping utilizes distributed routing algorithms such as Distributed Hash Table (DHT) to route data placement requests in an overlay network to reach the data destinations. Decentralized data mapping is used by many systems, such as Chord, Dynamo, and Cassandra, etc.
According to this scheme, a request to place particular data in the system is routed from storage node to storage node until the destination storage node is located. The next hop decisions used to route the request are based on a next hop algorithm accessible at each node. Specifically, when a particular node in the system receives the request to place the particular data, the node first checks whether it is the owner of the particular data (i.e., based on a set of ownership rules). If the particular node determines that it is not the owner of the particular data, then the node forwards the data placement request to one or more neighbor nodes, each of which determine whether it is the owner of the particular data by applying the same set of ownership rules as the particular node. Eventually, the data placement request reaches a storage node that determines that it is the owner of the particular data to be placed and this destination storage node stores the particular data.
Accordingly, data lookup is performed through routing the lookup request among the nodes of the system with the next hop being decided in a manner similar to the next hop decision for a request to place data on the system. A request for data is sent from node to node in the system until the owner of the requested data is located.
The number of hops to route the request is logarithmic to the number of nodes in the whole system, which is non-trivial for large-scale systems. Also, given that a query on data in a distributed query processing system may involve data units from a large number of nodes, the cumulative data lookup latency involved in responding to a query based on a decentralized scheme is non-trivial.
Therefore, it would be beneficial to develop an improved placement algorithm, for distributing data amongst the nodes of a distributed query processing system, that is an improvement over the centralized and decentralized systems described above.
The approaches described in this section are approaches that could be pursued, but not necessarily approaches that have been previously conceived or pursued. Therefore, unless otherwise indicated, it should not be assumed that any of the approaches described in this section qualify as prior art merely by virtue of their inclusion in this section.