CN118277456B - Initiating node output method in MPP distributed system - Google Patents

Abstract

The invention provides an originating node output method in an MPP distributed system, which relates to the technical field of databases, and comprises the following steps: collecting the number of nodes and the node IP in the database cluster, and sequencing according to the fragmentation information to obtain a first mapping relation of the node fragmentation and the node IP; collecting server information, creating an IP distance matrix of a computing node and a management node, and obtaining the distance matrix; distributing a plurality of hash buckets for each computing node, recording the corresponding relation between the computing node and the hash buckets, and acquiring a second mapping relation between the node fragments and the computing nodes according to nodedatamap table information and the corresponding relation; extracting hash keys in the filtering conditions of the query statement, and obtaining node IP of the computing node according to the hash key query; the distance matrix queries minimum values of the calculation node and the management node according to the node IP query result, and the corresponding node is output as the initiating node.

Description

A method for outputting initiating nodes in an MPP distributed system

Technical Field

The present invention relates to the field of database technology, and in particular to an initiating node output method in an MPP distributed system.

Background Art

In an MPP distributed system, it usually includes management nodes and computing nodes. The management node is the task initiating node, which is responsible for planning and scheduling tasks and receiving data returned from the computing nodes. The computing nodes are the nodes where the data is located, which are responsible for executing tasks and returning the data that needs to be returned to the initiating node. Therefore, the initiating node is equivalent to the chief dispatcher and plays an important role.

When the network load is not heavy and the number of concurrent tasks is not high, it does not seem to be critical to choose which management node to use as the initiating node for the task. However, in scenarios with heavy business load and high concurrency, the network bandwidth is squeezed and the cross-network communication network latency is high. The network can easily become a performance bottleneck, resulting in a backlog of tasks and the inability to complete the scenario business in a timely manner. At this time, it is crucial to correctly select the initiating node for the task.

In the past, DBAs usually used random selection or sequential selection to select task initiation nodes. However, neither of these two methods considered the impact of the network on performance, which prevented SQL from running on the best initiation node. This may cause performance anomalies due to network bottlenecks, affecting user usage.

Summary of the invention

The present invention aims to solve at least one of the technical problems existing in the related art. To this end, the present invention provides an initiating node output method in an MPP distributed system.

The present invention provides an initiating node output method in an MPP distributed system, comprising:

S1: Collect the number of nodes and node IPs in the database cluster, sort them according to the sharding information to obtain the first mapping relationship between the node shards and the node IPs;

S2: Collect server information, create an IP distance matrix between computing nodes and management nodes according to the server information, and obtain a distance matrix;

S3: Allocate multiple hash buckets to each computing node, record the correspondence between the computing nodes and the hash buckets through the nodedatamap table of the management node, and obtain the second mapping relationship between the node shards and the computing nodes according to the nodedatamap table information and the correspondence;

S4: extracting a hash key in the filter condition of the query statement, and obtaining the node IP of the computing node corresponding to the query statement according to the hash key query based on the second mapping relationship and the first mapping relationship, and obtaining a node IP query result;

S5: Obtain the minimum values of computing nodes and management nodes from the distance matrix according to the node IP query result, and output the node corresponding to the minimum value as the initiating node.

According to an initiating node output method in an MPP distributed system provided by the present invention, the first mapping relationship in step S1 is a one-dimensional array.

According to an initiating node output method in an MPP distributed system provided by the present invention, the distance matrix in step S2 is a two-dimensional matrix.

According to an initiating node output method in an MPP distributed system provided by the present invention, in the distance matrix, when the IP of the computing node is the same as the IP of the management node, the distance between the computing node and the management node is 0; when the IP of the computing node is different from the IP of the management node, the distance between the computing node and the management node is 1.

According to an initiating node output method in an MPP distributed system provided by the present invention, in step S3, when multiple hash buckets are allocated to each computing node, the allocation method is uniform allocation according to the total number of hash buckets.

According to an initiating node output method in an MPP distributed system provided by the present invention, the total number of hash buckets is 65536.

According to an initiating node output method in an MPP distributed system provided by the present invention, step S4 further comprises:

S41: extracting the hash value of the distribution column in the filter condition of the query statement;

S42: Obtain a hash key according to the hash value calculation;

S43: Obtaining the shard where the data corresponding to the query statement is located according to the hash key based on the second mapping relationship;

S44: Obtain the node IP corresponding to the computing node according to the shard where the data is located by using the first mapping relationship to obtain a node IP query result.

According to an initiating node output method in an MPP distributed system provided by the present invention, in step S4, the filtering condition is a precise query statement containing a where condition and having an equality distribution column.

The present invention provides an initiating node output method in an MPP distributed system, which is used to solve the problem of aggravated network load due to incorrect selection when the concurrency pressure is large and the network load is heavy. The method can automatically select the optimal initiating node according to the characteristics of MPP, ensure that the selected initiating node and the computing node are the same machine, save network overhead, avoid performance anomalies caused by incorrect selection of initiating nodes, enable user services to be completed more quickly, and improve user experience and usage efficiency.

Additional aspects and advantages of the present invention will be given in part in the following description and in part will be obvious from the following description, or will be learned through practice of the present invention.

BRIEF DESCRIPTION OF THE DRAWINGS

In order to more clearly illustrate the technical solutions in the present invention or the prior art, the drawings required for use in the embodiments or the description of the prior art will be briefly introduced below. Obviously, the drawings described below are some embodiments of the present invention. For ordinary technicians in this field, other drawings can be obtained based on these drawings without paying creative work.

FIG1 is a flow chart of an initiating node output method in an MPP distributed system provided by the present invention.

DETAILED DESCRIPTION

In order to make the purpose, technical scheme and advantages of the present invention clearer, the technical scheme of the present invention will be clearly and completely described below in conjunction with the drawings in the present invention. Obviously, the described embodiments are part of the embodiments of the present invention, rather than all the embodiments. Based on the embodiments in the present invention, all other embodiments obtained by ordinary technicians in the field without creative work are within the scope of protection of the present invention. The following embodiments are used to illustrate the present invention, but cannot be used to limit the scope of the present invention.

In the description of this specification, the description with reference to the terms "one embodiment", "some embodiments", "example", "specific example", or "some examples" etc. means that the specific features, structures, materials or characteristics described in conjunction with the embodiment or example are included in at least one embodiment or example of the embodiments of the present invention. In this specification, the schematic representations of the above terms do not necessarily refer to the same embodiment or example. Moreover, the specific features, structures, materials or characteristics described may be combined in any one or more embodiments or examples in a suitable manner. In addition, those skilled in the art may combine and combine the different embodiments or examples described in this specification and the features of the different embodiments or examples, without contradiction.

An embodiment of the present invention is described below with reference to FIG. 1 .

The present invention provides an initiating node output method in an MPP distributed system, comprising:

S1: Collect the number of nodes and node IPs in the database cluster, sort them according to the sharding information to obtain the first mapping relationship between the node shards and the node IPs;

Wherein, the first mapping relationship in step S1 is a one-dimensional array.

In this stage, we first obtain the mapping relationship between shards and IP addresses. Specifically, we obtain the number of nodes and IP addresses in the database cluster and sort them by shard information. We define an N-column one-dimensional array A, where N is the number of computing nodes. The expression is A[0]=ip1, A[1]=ip2…A[N-1]=ipN.

S2: Collect server information, create an IP distance matrix between computing nodes and management nodes according to the server information, and obtain a distance matrix;

Wherein, the distance matrix in step S2 is a two-dimensional matrix.

In the distance matrix, when the IP of the computing node is the same as the IP of the management node, the distance between the computing node and the management node is 0; when the IP of the computing node is different from the IP of the management node, the distance between the computing node and the management node is 1.

The purpose of this stage is to obtain the distance matrix between the computing node and the initiating node. Specifically, first define an N*N two-dimensional matrix M based on the server location information or IP information provided by the server administrator, and then create an IP distance matrix between the computing node and the management node. If the IPs are the same, the distance is 0, and if the IPs are different, the distance is 1.

S3: Allocate multiple hash buckets to each computing node, record the correspondence between the computing nodes and the hash buckets through the nodedatamap table of the management node, and obtain the second mapping relationship between the node shards and the computing nodes according to the nodedatamap table information and the correspondence;

Furthermore, the purpose of this stage is to obtain the relationship between shards and computing nodes. The MPP cluster has N computing nodes, with ids ranging from n1 to nN. Each node is allocated 65536/N hash buckets, and the hash bucket ids are h0-h65535.

The nodedatamap (system table used to record the mapping between node shards and data) in the MPP initiating node records the relationship between shards and hash buckets. Each node corresponds to multiple hash buckets. Then we can obtain the relationship between the shards and the computing node IP through the information in the nodedatamap.

In step S3, when multiple hash buckets are allocated to each computing node, the allocation method is to evenly allocate according to the total number of hash buckets.

Among them, the total number of hash buckets is 65536.

In some embodiments, the GBase database corresponds to 65536 hash buckets.

S4: extracting a hash key in the filter condition of the query statement, and obtaining the node IP of the computing node corresponding to the query statement according to the hash key query based on the second mapping relationship and the first mapping relationship, and obtaining a node IP query result;

Wherein, step S4 further comprises:

S41: extracting the hash value of the distribution column in the filter condition of the query statement;

S42: Obtain a hash key according to the hash value calculation;

S43: Obtaining the shard where the data corresponding to the query statement is located according to the hash key based on the second mapping relationship;

S44: Obtain the node IP corresponding to the computing node according to the shard where the data is located by using the first mapping relationship to obtain a node IP query result.

Wherein, in step S4, the filtering condition is a precise query statement containing a where condition and having an equal distribution column.

In this stage, first extract the hash value v in the filter condition in SQL and determine the IP of the SQL computing node. Specifically, after obtaining SQL, find the value v of the hash distribution column in the precise query filter condition. The sign is that it contains the where condition and the distribution column '=' exists. After finding the value v, perform crc32(v)%65536 to calculate the hashkey, where crc is an encryption and decryption algorithm based on the crc verification principle. Then, obtain the relationship between the shard and the hashkey from the nodedatamap information in step S3 to obtain the shard where the data is located. According to the shard and IP mapping relationship obtained in step S1, the computing node IP is obtained.

S5: Obtain the minimum values of computing nodes and management nodes from the distance matrix according to the node IP query result, and output the node corresponding to the minimum value as the initiating node.

The purpose of this stage is to obtain the optimal initiating node IP. According to the computing node IP obtained in step S4, the minimum value of the distance matrix between the computing node and the initiating node obtained in step S2 is found, min(M[k,i]), k is the shard value of the computing node IP, i is 0...N-1, and the M node is output as the computing node.

Finally, it should be noted that the above embodiments are only used to illustrate the technical solutions of the present invention, rather than to limit it. Although the present invention has been described in detail with reference to the aforementioned embodiments, those skilled in the art should understand that they can still modify the technical solutions described in the aforementioned embodiments, or make equivalent replacements for some of the technical features therein. However, these modifications or replacements do not deviate the essence of the corresponding technical solutions from the spirit and scope of the technical solutions of the embodiments of the present invention.

Claims (7)

1. An originating node output method in an MPP distributed system, comprising:

s1: collecting the number of nodes and the node IP in the database cluster, and sequencing according to the fragmentation information to obtain a first mapping relation of the node fragmentation and the node IP;

s2: collecting server information, creating an IP distance matrix of a computing node and a management node according to the server information, and obtaining a distance matrix;

s3: a plurality of hash buckets are distributed for each computing node, the corresponding relation between the computing node and the hash buckets is recorded through a nodedatamap table of a management node, and a second mapping relation between the node fragments and the computing node is obtained according to nodedatamap table information and the corresponding relation, specifically comprising:

S31: assigning a plurality of hash buckets to each computing node;

s32: recording the corresponding relation between the computing node and the hash bucket through nodedatamap tables of the management nodes;

S33: recording the corresponding relation between the node fragments and the hash bucket through a nodedatamap table of the MPP initiating node;

S34: obtaining a hash bucket corresponding to a computing node through a nodedatamap table of a management node, obtaining node fragments corresponding to the hash bucket through a nodedatamap table of an MPP initiating node, and obtaining a second mapping relation between the node fragments and the computing node according to the hash bucket corresponding to the computing node and the node fragments corresponding to the hash bucket;

S4: extracting a hash key in a filtering condition of a query statement, and obtaining a node IP of a computing node corresponding to the query statement according to the hash key query by using the second mapping relation and the first mapping relation to obtain a node IP query result, wherein the method specifically comprises the following steps of:

S41: extracting hash values of the distributed columns in the filtering conditions of the query statement;

s42: calculating according to the hash value to obtain a hash key;

s43: obtaining node fragments corresponding to hash keys through a nodedatamap table of an MPP initiating node according to the second mapping relation, and obtaining the fragments corresponding to the query statement data according to the hash keys;

S44: acquiring node IP corresponding to a computing node according to the fragments of the data by the first mapping relation, and acquiring a node IP query result;

s5: and inquiring and obtaining minimum values of the calculation node and the management node according to the node IP inquiry result by the distance matrix, and outputting the node corresponding to the minimum value as an initiating node.

2. The method as claimed in claim 1, wherein the first mapping relation in step S1 is a one-dimensional array.

3. The method according to claim 1, wherein the distance matrix in step S2 is a two-dimensional matrix.

4. The method as set forth in claim 3, wherein in the distance matrix, when the IP of the computing node is the same as the IP of the management node, the distance between the computing node and the management node is 0, and when the IP of the computing node is different from the IP of the management node, the distance between the computing node and the management node is 1.

5. The method as set forth in claim 1, wherein in step S3, when a plurality of hash buckets are allocated to each computing node, the allocation is performed uniformly according to the total number of hash buckets.

6. An originating node output method in an MPP distributed system as defined in claim 5, wherein the hash buckets total 65536.

7. The method as set forth in claim 1, wherein in step S4, the filtering condition is an exact query statement containing a where condition and having an equation distribution column.