Cluster-node load balancing in a distributed database system

In one exemplary aspect, a method of a cluster-node load balancing system of a distributed database system includes receiving a request from a cluster with at least one node of a cluster of the distributed database system. The request includes a query for an identity of all other nodes known by the node as well as a metadata of all data maintained by the node. The identity of all other nodes known by the node as well as the metadata of all data maintained by the node is provided to the cluster.

CROSS-REFERENCE TO RELATED APPLICATIONS

This application incorporates by reference the following U.S. patent applications: U.S. patent application Ser. No. 14/299,566, titled METHOD AND SYSTEM OF MAPREDUCE IMPLEMENTATIONS ON INDEXED DATASETS IN A DISTRIBUTED DATABASE ENVIRONMENT and filed on Jun. 9, 2014 U.S. Pat. No. 9,002,871; U.S. patent application Ser. No. 13/653,411, titled METHOD AND SYSTEM OF MAPREDUCE IMPLEMENTATIONS ON INDEXED filed on Oct. 17, 2012 U.S. Pat. No. 8,775,464; U.S. application Ser. No. 13/451,551, titled REAL-TIME TRANSACTION SCHEDULING IN A DISTRIBUTED DATABASE and filed Apr. 20, 2012 U.S. Pat. No. 8,799,248; and U.S. Provisional Application No. 61/478,940, titled DISTRIBUTED DATABASE SYSTEM WITH A CLUSTER OF AUTONOMOUS NODES and filed Apr. 26, 2011. These applications are hereby incorporated by reference in their entirety.

BACKGROUND

This application relates generally to data storage, and more specifically to a system, article of manufacture and method of cluster-node load balancing in a distributed database system.

2. Related Art

A distributed database can include a plurality of database nodes and associated data storage devices. A database node can manage a data storage device. If the database node goes offline, access to the data storage device can also go offline. Accordingly, redundancy of data can be maintained. However, maintaining data redundancy can have overhead costs and slow the speed of the database system. Additionally, offline data may need to be rebuilt (e.g. after the failure of the database node and subsequent rebalancing operations). This process can also incur a time and processing cost for the database system. Therefore, methods and systems of self-managing nodes of a distributed database cluster with a consensus algorithms can provide improvements to the management of distributed databases.

BRIEF SUMMARY OF THE INVENTION

In one aspect, a method of a cluster load balancing system of a distributed database system includes receiving a request from a cluster with at least one node of a cluster of the distributed database system. The request includes a query for an identity of all other nodes known by the node as well as a metadata of all data maintained by the node. The identity of all other nodes known by the node as well as the metadata of all data maintained by the node is provided to the cluster.

In another aspect, a method of a distributed database system includes providing a first server node of a node cluster of a Not Only SQL (NoSQL) distributed database system. The first server node maintains a node address list of a set of active cluster nodes of the node cluster. A second cluster node of the node cluster of the NoSQL distributed database system is provided. Upon joining the node cluster, the second cluster node is configured with an address of the first server node. The second cluster node requests the node address list from the first server node. The second cluster node polls each individual node of the set of active cluster nodes to determine the portions of the data of the NoSQL distributed database system that each individual node maintains.

Optionally, the first server node and the second cluster node communicate using an ASCII-based control protocol. The first server node communicates the node address list to the second cluster list after the second cluster node requests the node address list from the first server node. The node address list comprises an identity the set of active cluster nodes of the node cluster and a metadata of all data maintained by the first node. Each individual node responds to the second node with the metadata comprising a distributed hash table comprising the data maintained by each individual node. The node cluster is designed without a node master and without database sharding. The second node creates a data map of set of active cluster nodes of the node cluster, and wherein the data map is used to populate the node address list in the second node.

DETAILED DESCRIPTION

Disclosed are a system, method, and article of manufacture for cluster-node load balancing in a distributed database system. The following description is presented to enable a person of ordinary skill in the art to make and use the various embodiments. Descriptions of specific devices, techniques, and applications are provided only as examples. Various modifications to the examples described herein may be readily apparent to those of ordinary skill in the art, and the general principles defined herein may be applied to other examples and applications without departing from the spirit and scope of the various embodiments.

A. Environment and Architecture

Disclosed are a system, method, and article of manufacture of a distributed database system.FIG. 1shows, in a block diagram format, a distributed database system (DDBS) operating in a computer network according to an example embodiment, according to some embodiments. A DDBS can typically be a collection of databases that can be stored at different computer network sites. Each database may involve different database management systems and different architectures that distribute the execution of transactions. A DDBS can be managed in such a way that it appears to the user as a centralized database.

FIG. 1shows database nodes104A-C (hereafter ‘nodes’) that collectively represent a cluster of computers108and associated databases (hereafter ‘cluster’) usable by a DDBS. The processing and data storage of the DDBS can be partitioned into nodes104A-C. The term “cluster” refers to a collection of one or more networked database nodes that function as a single system. In one example embodiment, cluster108can be designed without a node master and without database sharding. Database sharding typically involves a method of horizontal partitioning in a database or search engine. The architecture of cluster108can establish node equality and use uniform data distribution across cluster108. Cluster108can also utilize a per node structure in order to provide continuity of processing cluster data transactions in the event of a single point of failure within the cluster. Additionally, cluster108can use a data rebalancing mechanism that can evenly distribute query volume across all nodes in a manner that rebalancing does not generally affect the behavior of cluster108. Cluster108can also automatically handle network-partitioning events. These operations are discussed in further detail below.

In a particular example embodiment, cluster108can be implemented with a shared-nothing architecture. A shared-nothing architecture is typically characterized by data partitioning and no sharing between the machine components in a cluster of computers, except where communication between partitions is carried out. The database task carried out by the cluster is subdivided and each machine carries out processing steps using its own resources to complete its subdivided portion or portions of the task. Such a cluster architecture can scale for database workloads and have a limited need for intra-cluster communication.

Nodes104A-C can communicate to clients100A-N via IP network102. Internet-protocol (IP) network102can utilize a set of communications protocols used for the Internet and other similar networks. In some embodiments, IP network102may also include other means of data communication such as a local area network (LAN) that utilizes IEEE 802-type protocols, a telecommunications data network, or any combination thereof. Clients100A-N can be any application or process that communicates with nodes104A-C via IP network102.

Nodes104A-C can include one or more central processing units (CPU), memory resources and permanent data storage systems. Database nodes104A-C can include distributed database management system (DDBMS)106A-C. DDBMS106A-C can include a set of computer programs that controls the creation, maintenance, and the use of distributed database of cluster108. DDBMS106A-C manages the various data storage systems114,118and120that comprise the distributed database as well as the data objects on the data storage systems. The particular example embodiment ofFIG. 1shows DDBMS106A-C as a distributed database manager layer. The DDBMS106A-C can include components that are able to execute database tasks on their respective servers110A-C, as well as to carry out functions (described below) that relate to the operation of the DDBS in cluster108.

At the application layer of the database nodes104A-C can manage the processing of data transactions. For the sake of simplicity, not all the components of nodes104A-C are shown. However, it will be appreciated that nodes104A-C can include other components. For example, DDBMS106A-C can include systems as lock managers, schedulers, metadata managers, policy managers and the like. In some embodiments, nodes104A-C can be self-managing nodes that can reconfigure the cluster and repartition data within the cluster without a central administrative entity such as a database administrator (DBA).

Nodes104A-C can be linked together via an internal cluster interconnect124such as a Fibre Channel network. Fibre Channel protocols can use a transport protocol (similar to TCP used in IP networks) which predominantly transport small computer system interface (SCSI) commands over a Fibre Channel network. SCSI commands are typical a set of standards for physically connecting and transferring data between computers and peripheral devices. In other embodiments, internal cluster interconnect124can use internet small computer system interface (iSCSI) protocols as well. iSCSI can carry SCSI commands over (and thus link nodes104A-C via) an IP network.

A database transaction can comprise a unit of work performed within the data storage system (or similar system) against a database, and is treated in a coherent and reliable way generally discreet of other data transactions. Generally, a database transaction has four properties that lead to the consistency and reliability of a distributed database. These are Atomicity, Consistency, Isolation, and Durability.

A transaction is treated as a unit of operation. For example, in the case of a crash, the system should complete the remainder of the transaction, or it will undo all the actions pertaining to this transaction. Should a transaction fail, changes that were made to the database by it are undone (i.e. rollback).

This property deals with maintaining consistent data in a database system. A transaction can transform the database from one consistent state to another. Consistency falls under the subject of concurrency control.

Each transaction should carry out its work independently of any other transaction that may occur at the same time.

This property ensures that once a transaction commits, its results are permanent in the sense that the results exhibit persistence after a subsequent shutdown or failure of the database or other critical system. For example, the property of durability ensures that after a COMMIT of a transaction, whether it is a system crash or aborts of other transactions, the results that are already committed are not modified or undone.

Additionally, nodes104A-C can also include of one or more data storage devices114,118and120(e.g. a dynamic random-access memory (DRAM), rotating hard disk, solid-state drive (SSD), or any combination thereof) or file systems. Each data storage device114,118and120can have a set ofinetadata that uniquely identifies the data its stores and the data attributes (e.g. time stamps and the like) to the DDBMS that manages the particular storage device.

An SSD device can be a data storage device that uses solid-state memory to store persistent data with the intention of providing access in the same manner of a block input/output hard disk drive. DRAM can be a type of random-access memory that stores each bit of data in a separate capacitor within an integrated circuit. The capacitor can be either charged or discharged; these two states are taken to represent the two values of a bit (0 and 1). A hard-disk drive can be a non-volatile magnetic data storage device in which data are digitally recorded by various electronic, magnetic, optical, or mechanical methods on a surface layer deposited of one or more planar, round and rotating platters.

FIG. 2depicts a block diagram of an exemplary database platform that can be implemented in a DDS such as the system ofFIG. 1, according to some embodiments. Database platform200includes both hardware architecture and software frameworks that allow the database systems, such as the software functionalities of the query layer202, the distribution layer204and the data storage layer206, to operate.

The query layer202can include the client libraries and the query mechanisms. Client libraries can include smart client libraries, including libraries in the PHP, Java, C#, C, C libevent, Python and Ruby on Rails languages. Query layer202can include systems and functionalities that support various query types from clients100A-N. Query layer202can handle client communications in various protocol formats such as an ASCII-based control protocol. This protocol can govern the client's server discovery process and the client-server handshake. Query layer202can be optimized for key-value queries as well as other query types. Additional information regarding the query layer202is provided below in the description ofFIG. 4.

Distribution layer204can include systems and functionalities that implement and manage a distributed node cluster architecture (such as those described above with regards to the system ofFIG. 1) that can combine distributed transactions with server distribution. Distribution layer204can implement such operations as inter-cluster communication, cluster-consensus voting operations, namespace distribution, distributed transaction management, replication operations and maintenance operations.

Data storage layer206can include systems and functionalities that support a variety of data models according to the various embodiments. In one example embodiment, the data storage layer206can include three functional units (not shown). In this example embodiment, the data model can provide application semantics that include named columns and typed values. A primary key index can be used to perform data lookup, data retrieval and data iteration operations. The systems and functionalities of the data storage layer206can also utilize various data storage systems, including DRAM, rotational disk, flash storage, or any combination thereof (e.g. data on rotational disk using available DRAM as a cache). Flash storage can be implemented with an SSD device. Furthermore, in this example embodiment, the data storage layer206can implement a schema-free data model that supports the standard read/write operations and additionally supports the ability to increment values within the distributed database. The data storage layer206can additionally implement indexes that are stored in DRAM.

Data storage layer206can spread the contents of each namespace across every node in a cluster. This virtual partitioning can be automatic and transparent to a client. If a node receives a request for a piece of data it does not have locally, the node can then satisfy the request by creating an internal proxy for this request. The node can then fetch the data from the real owner node and then subsequently reply to the client directly. It should be noted that other aspects of database platform200(e.g. a data transport layer) have not been shown for the sake of simplicity.

FIG. 3depicts an exemplary computing system300configured to perform any one of the processes described herein. In this context, computing system300may include, for example, a processor, memory, storage, and I/O devices (e.g., monitor, keyboard, disk drive, Internet connection, etc.). However, computing system300may include circuitry or other specialized hardware for carrying out some or all aspects of the processes. In some operational settings, computing system300may be configured as a system that includes one or more units, each of which is configured to carry out some aspects of the processes either in software, hardware, or some combination thereof.

FIG. 3is a block diagram illustrating a computing system300, according to some embodiments. The computing system300is based upon a suitably configured processing system adapted to implement one or more exemplary embodiments. Any suitably configured processing system can similarly be used as the computing system300by embodiments such as servers110A-C residing in cluster108ofFIG. 1, a personal computer, workstation, a distributed database server, or the like. The computing system300includes a computer302. The computer302has a processor(s)304that is connected to a memory306, mass storage interface308, terminal interface310, and network adapter hardware312. A system bus314interconnects these system components. The mass storage interface308is used to connect mass storage devices, such as data storage device316(e.g. data storage systems114,118and120and data storage708described infra), to the computer302. Examples of data storage316can include those examples discussed supra (rotating hard disk systems, SSD flash systems, DRAM, and the like), as well others such as optical drives. Data storage316may be used to store data to and read data from a computer-readable medium or storage product.

Memory306, in one embodiment, includes a DDBMS, such as DDBMS106A-C. In some example embodiments, memory306can also include one or more indexes. Although illustrated as concurrently resident in the memory306, it is clear that respective components of the memory306are not required to be completely resident in the memory306at all times or even at the same time. In one embodiment, the computer302utilizes conventional virtual addressing mechanisms to allow programs to behave as if they have access to a large, single storage entity, referred to herein as a computer system memory, instead of access to multiple, smaller storage entities such as the memory306and data storage device316. In some embodiments, additional memory devices (such as a DRAM cache) can be coupled with computer302as well.

Although only one CPU304is illustrated for computer302, computer systems with multiple CPUs can be used equally effectively. Some embodiments can further incorporate interfaces that each includes separate, fully programmed microprocessors that are used to off-load processing from the CPU304. Terminal interface310is used to directly connect one or more terminals320to computer302to provide a user interface to the computer302. These terminals320, which are able to be non-intelligent or fully programmable workstations, are used to allow system administrators and users to communicate with computer302. The terminal320can also include other user interface and peripheral devices that are connected to computer302and controlled by terminal interface hardware included in the terminal I/F310that includes video adapters and interfaces for keyboards, pointing devices, and the like.

An operating system (not shown) included in the memory is a suitable multitasking operating system such as the Linux, UNIX, Windows XP, and Windows Server operating system. Embodiments are able to use any other suitable operating system. Some embodiments utilize architectures, such as an object oriented framework mechanism, that allows instructions of the components of operating system to be executed on any processor located within computer302. The network adapter hardware312is used to provide an interface to a network322. Some embodiments are able to be adapted to work with any data communications connections including present day analog and/or digital techniques or via a future networking mechanism.

Although the exemplary embodiments are described in the context of a fully functional computer system, those skilled in the art will appreciate that embodiments are capable of being distributed as a program product via CD or DVD, e.g., a CD ROM, or other form of recordable media, or via any type of electronic transmission mechanism. At least some values based on the results of the above-described processes can be saved for subsequent use. Additionally, a computer-readable medium can be used to store (e.g., tangibly embody) one or more computer programs for performing any one of the above-described processes by means of a computer. The computer program may be written, for example, in a general-purpose programming language (e.g., Pascal, C, C++, and Java) or some specialized application-specific language.

FIGS. 4A-B depicts an example cluster load balancing system of a DDBS that utilizes the query layer202ofFIG. 2supra, according to some embodiments. Client100can communicate to nodes104A-C using an ASCII-based control protocol. The control protocol can govern such operations as the client's server discovery process and the client server handshake. Upon startup of cluster108, client100can be configured with the address of at least one of nodes104A-C. Once client100has secured an address, client100can then request the addressed node for the remaining node addresses. In this way, client100can create a node address list400. Address list400can be periodically maintained and updated. Additionally, client100can poll each individual node to determine the portions of the data of the DDBS that the particular node maintains. Nodes104A-C then respond with metadata (e.g. a distributed hash table (DHT)) about the data maintained by the particular node. Client100and nodes104A-C can use a client-side metadata-communication protocol to communicate metadata queries and responses between each other. Use of the metadata-communication protocol allows for client/server data exchanges to proceed in a substantially concurrent manner with the operations ofFIGS. 4A-B. In this way, client100can then create and maintain a data map of nodes104A-C. The data map can be used to populate address list400and/or client libraries402(see below).

Client libraries402can include the source code in a computer programming language such as Java, PHP, C#, C, C libevent, Python and Ruby on Rails. Client libraries402can include both blocking and event-oriented interfaces. Client applications can link against the Client libraries402. In this way, client libraries402can expose a common set of interfaces for data insertion, retrieval, modification, and deletion. Client libraries402can also connect a client application to cluster108. Client libraries402can implement a discovery protocol and route each query to nodes104A-C (or, in some cases, determine and route the query to an optimal node). Client libraries402can include information as to where individual data elements are stored. Client libraries402dynamically tracks the size and state of cluster108, so no reconfiguration is necessary when cluster nodes are added or removed. Additionally, client libraries402can automatically retry transactions safely in failure cases, or if desired, can apply transactions only once. Client libraries402can include mechanisms for retrying writes and using a client-generated unique persistent transaction identifier.

In one example embodiment, client100can internally follow the cluster with the following operation. Client100can request that a node (such as node104A) generate a list of peer (‘friend’) nodes (e.g.104B and C) in cluster108and determine which data is maintained by each peer node. Nodes104A-C can communicate this information via internal interconnect124. Thus, node104A would return a list of nodes104B and C and their respective concomitant data. This information can then be provided to client100. Client100can then iterate this operation with each peer node provided in a list from node104(i.e. repeat the process with node104B and then104C). In this way, client100can discover any peer nodes that may not be known to another node. For example, a node104D (shown inFIG. 4B) may be included in the cluster. Node104A may not be aware of this node due to some error in cluster reconfiguration. However, node104B may be aware of node104D. As a result, as client100iterates the operation, it will become aware of node104D and its concomitant data. Thus, client100can develop a complete list of all the nodes in cluster108. The list can include metadata that provides the location of the data in cluster108as well. The list can be included in client libraries402. The list of nodes104A-C and104D can be included in address list400.

FIG. 5depicts an example process500, according to some embodiments. Step502includes receiving a request from at least one node of a cluster of the distributed database system, wherein the request comprises a query for an identity of all other nodes known by the node as well as a metadata of all data maintained by the node. Step504includes providing the identity of all other nodes known by the node as well as the metadata of all data maintained by the node to the at least one node.

FIG. 6illustrates an example process600of a distributed database system according to some embodiments. Step602includes providing a first server node of a node cluster of a Not Only Structured Query Language (NoSQL) distributed database system, wherein the first server node maintains a node address list of a set of active cluster nodes of the node cluster. Step604includes providing a second cluster node of the node cluster of the NoSQL distributed database system, wherein upon joining the set of active cluster nodes, the second cluster node is configured with an address of the first server node, wherein the second cluster node requests the node address list from the first server node, wherein the second cluster node polls each individual node of the set of active cluster nodes to determine the portions of the data of the NoSQL distributed database system that each individual node maintains.

CONCLUSION