CN104809210B - One kind is based on magnanimity data weighting top k querying methods under distributed computing framework - Google Patents

Abstract

The invention discloses a top-k query optimization method for massive data based on a spark distributed computing framework, which divides massive data sets in advance, mainly using a grid-like data segmentation method. Divide the original data set into different sub-data sets, and then select a small number of suitable sub-data sets to replace the entire data set for query according to the weight given by the user to each attribute of the data object and the query k value. Experimental results prove that the method proposed in this paper has a fast query speed and good scalability. Compared with the traditional top-k query method and the data segmentation method based on angle and distance, the query speed is improved, and the information that the user needs to query can be fed back in a short time.

Description

A mass data weighted top-k query method based on distributed computing framework

technical field

The invention relates to a data query method, in particular to a weighted top-k query method in massive data sets.

Background technique

Top-k query, also known as rank-aware query, is one of the most basic operations in the database, and it is also an important tool for data analysis. Especially in business analysis, it is often only necessary to focus on the most useful data, while Not the entire dataset.

The definition of top-k query is as follows: use D={T ₁ ,T ₂ ,…,T _n } to represent the collection of all data objects, T _i represents the i-th data object, each data object has d dimensions, and all is a point in space. For a top-k query Q(f,k), f represents the scoring function, and k represents returning k results that meet the query requirements. f is a weighted sum function, that is, for a data object T(t ₁ ,t ₂ ,…,t _d ) in the sample, the user assigns a weight W(w ₁ ,w ₂ ,…, w _d ), the score of each data object is obtained by the weighted sum of each attribute value, that is, the score function is:

The top-k query only needs to get a result set of k elements at the end, and it can be obtained by sorting the data that is much smaller than the input data set, without processing the global data. In recent years, with the explosive growth of data scale, massive data scale has brought great challenges to data storage, management and analysis. As a basic operation in data analysis, top-k query needs to get query results quickly. For example: in Taobao's mass products, users assign different weights to product attributes according to their own preferences, and then the system quickly returns the top k products that meet user needs according to user needs.

However, top-k queries on massive data face two challenges: one is that the data scale reaches TB or PB level, and traditional centralized data processing methods are no longer applicable; the other is how to quickly and accurately obtain query results for massive data sets.

Top-k queries in traditional centralized data systems encounter performance bottlenecks in massive data, so they are not suitable for processing massive data sets. In the traditional distributed environment, some studies improve the query efficiency by caching the query results, but this method does not solve the problem of massive data top-k query in essence; some use the Skyline outline query method for data processing, A complete set of top-k processing framework for DiTo is proposed, but it is only in a traditional distributed environment.

In recent years, the most basic solution to the top-k problem in the cloud environment is to sort all the data and return the top k results, but this method needs to process the original data set for each query, resulting in redundant workload. The query time is long, so it is not advisable. RanKloud et al. proposed to calculate the threshold for early termination of queries through statistical analysis of system runtime under the MapReduce framework. This method cannot guarantee k accurate results. There are also studies that use the cache mechanism to compare the similarity between a new query and the existing query in the cache. If the similarity is large, there is no need to re-query. Although the query speed is accelerated, the query results are not accurate. It has been proposed to divide top-k queries based on angle and distance data, but the angle-based data division scheme is complex and time-consuming to convert data coordinates, so it is not suitable for processing massive data sets.

Contents of the invention

Purpose of the invention: In order to overcome the deficiencies in the prior art, the present invention provides a massive data weighted top-k query method based on a distributed computing framework, which is used to solve the problem that the existing top-k query cannot be processed quickly when processing massive data. , Accurate and easy access to technical issues of query results.

Technical scheme: in order to achieve the above object, the technical scheme adopted in the present invention is:

First make the following four reasonable assumptions:

(1) Any attribute value of a data object takes a non-negative value, and even a negative value can be converted into a non-negative value through normalization of the data.

(2) The data set is relatively fixed, or the update speed of the data is negligible within a certain period of time compared to the entire data set. little change over a period of time. Therefore, for stream data processing, the method of the present invention is not applicable.

(3) The data is evenly distributed in space. In massive data sets, this assumption is consistent in many scenarios.

(4) For an input weight W, satisfy Even if it is not, it can be obtained by normalization.

On the basis of the above four reasonable assumptions, a weighted top-k query method based on massive data under the distributed computing framework is proposed, including the following steps executed sequentially:

Step 1. Create a data space

Firstly, the attribute values of all data objects containing d attributes are converted into non-negative values, and the attribute values are normalized; a d-dimensional coordinate system is established, and the axes of the coordinate system correspond to the attributes of the data objects one by one. All data objects are placed in the coordinate system to form a data space;

Step 2. Data division

Starting from the origin of the coordinate system, the entire coordinate system is divided into m areas from the inside to the outside. The value of m here cannot be too large, otherwise it will bring adverse consequences of increased calculation. In the case of the current data scale, Generally, m is taken as 3 to 5. This range reasonably considers the data scale and calculation amount. Of course, as the data scale further increases in the future, the value of m can be appropriately increased to obtain the data in the divided area. The purpose of reducing the total amount; each area is numbered 1, 2, ..., m from the outside to the inside, and the boundary of the first area and the coordinate axis together include all data objects. For any area, The maximum value of each property of this area is the same, and the coordinates of the outer boundary of each area satisfy at least one axis whose coordinate value is the maximum value of the property of this area, and the maximum value of the property of the first area is set to a Under the premise of ₁ , the maximum value of the attribute of the i-th area i=1,2,...,m. After the regions are divided according to the above method, combined with the assumption (3) above, it can be known that the amount of data in each region is equal.

Except for the outermost area, for each of the remaining areas, the point that belongs to the area and the attribute of each axis is the maximum value is used as the base point, and all the attribute values in the entire coordinate system are greater than or equal to the corresponding attribute value at the base point. The regions are all divided, numbered 1, 2, ..., M in order from outside to inside, where M=m-1, and the above newly divided regions are used as judgment regions.

According to the principle of Skyline, for any two points 1 and 2, if all attribute values of point 1 are smaller than point 2, then point 1 supports point 2. Based on the above principles, if two data objects T ₁ and T ₂ are given, if for The attribute value of T ₁ is greater than or equal to the corresponding attribute value of T ₂ ie T ₁ .t _i ≥ T ₂ .t _i , t _i represents the attribute value of the i-th attribute, then an input weight W(w ₁ ,w ₂ ,…,w _d ), there must be a score of T ₁ greater than that of T ₂ , that is, f _W (T ₁ )≥f _w (T ₂ ).

Based on the above analysis, the scores of data objects in a judgment area must be greater than the scores of data objects in all areas inside the area to which this judgment area belongs. Since the results returned by the algorithm are taken from the k with the highest scores, k is the number of data objects in the result set returned by the algorithm, so once the number of data objects in a certain judgment area is greater than or equal to k, then these k data objects must not be from the inner area of the area to which the judgment area belongs acquired. Therefore, based on the above analysis, the judgment area is judged as follows:

According to the order of numbering from small to large, judge whether N _i ≥ k holds true, where N _i is the number of data objects in the i-judgment area, and k is the number of data objects in the result set returned by the algorithm; when i judges If the area satisfies the above formula and holds, the judgment ends, and the i areas from outside to inside are used as search areas for searching.

Further, in the present invention, the innermost area in the search area is subdivided, the area number is i, and the subdivision method is as follows:

Divide the area into d+1 blocks, where block d is the search area to be selected, and the remaining areas except the search area to be selected are mandatory search areas;

Number the search areas to be selected as n=1, 2,...,d, where any data point T _nj (t _n1 ,t _n2 ,...,t _nd ) in the nth search area to be selected, where t _nj Indicates the attribute value corresponding to the j-axis of the data point T _nj , t _nj satisfies the following two inequalities:

0≤t _nj ≤2a _i+1 -a _i , where 1≤j≤d and j≠n (1)

a _i -a _i+1 ≤t _nj ≤a _i , where j=n (2)

In the nth search area to be selected, if the data point T _nj (t _n1 ,t _n2 ,…,t _nd ) satisfies that the attribute value corresponding to one axis is a _i and the attribute value corresponding to the other axes is 2a _i+1 - a _i , then use this data point as the maximum boundary point of the nth candidate retrieval area;

Traversing whether each attribute weight w _j at the maximum boundary point of each candidate retrieval area assigned by the user satisfies the following conditions:

If there is an attribute weight w _j of the maximum boundary point that satisfies the above conditions, the scope of the search area is narrowed to include i-1 areas from outside to inside, the required search area in the i-th area, and the candidate to be selected to which the maximum boundary point belongs search area;

If there is no attribute weight w _j of the maximum boundary point that satisfies the above conditions, the search area is narrowed to include i-1 areas from outside to inside and the required search area in the i-th area.

On the premise that the judgment area N _i ≥ k is established, the subdivision method divides the innermost search area into the required search area and the candidate search area, and further selects the appropriate part from the candidate search area according to the judgment principle Search to further narrow the search scope. According to the previous argument, the score at the maximum boundary point of each search area to be selected must be greater than or equal to the score of data points in other positions in the search area to be selected; therefore, if the maximum boundary point of a search area to be selected If the score is less than the score at the base point of the judgment area N _i , then the search area to be selected does not need to be searched; otherwise, the search area to be selected needs to be searched.

For the convenience of illustration, select a search area to be selected, and the coordinates of its maximum boundary point are T(a _i ,2a _i+1 -a _i ,2a _i+1 -a _i ,…,2a _i+1 -a _i ), The base point coordinates of the corresponding judgment area N _i are T(a _i+1 ,a _i+1 ,a _i+1 ,...,a _i+1 ); substitute the above coordinates into the score function, if (a _i , 2a _i+1 -a _i ,2a _i+1 -a _i ,…,2a _i+1 -a _i )*W＞(a _i+1 ,a _i+1 ,…,a _i+1 )*W, Here W=(w ₁ ,w ₂ ,…,w _d ), then the above formula can be transformed into From this you can get Then it is necessary to search the region to be retrieved corresponding to the above maximum boundary point; according to the above method, judge all the regions to be retrieved, and a unified expression can be obtained Here it can be further concluded that if a certain area to be retrieved satisfies the condition, then the attribute weight w _j that makes the inequality true must be the weight of the attribute whose attribute value is a _i at the largest boundary point of the area to be retrieved, then when traversing the search, As long as the weight of the attribute whose attribute value is a _i of the maximum boundary point in each area to be retrieved is brought into If it is true, the area to be retrieved needs to be retrieved, and once an attribute weight is found that makes the above formula true, there is no need to continue to check other areas to be retrieved, because the attribute weights have been normalized, so it is impossible to have 2 or More than 2 attribute weights satisfy the above inequality; therefore, when selecting the area to be retrieved, you can quickly judge whether the largest attribute weight is full according to the attribute weight value input by the user If it is satisfied, then correspondingly find out the area to be retrieved whose jth attribute of the maximum boundary point is a _i .

Beneficial effect:

The present invention provides a mass data weighted top-k query method based on a distributed computing framework, and proposes a new data segmentation method similar to grid division, and judges the difference between the amount of data in the area and the amount of data k in the result set How many comparisons are made to preliminarily determine the search area, which greatly narrows the search range; then further narrows the search range of the innermost search area, making the final search area smaller and improving search efficiency and speed.

According to the statistical results, the data space with 1 billion pieces of data is divided into m=3 areas, and the amount of data in the outermost judgment area varies with the dimension d, that is, the number of attributes contained in each data object, as shown in the following table trend:

Table 1

It can be seen from the table that when the dimension d is less than 8, there are still 18 data objects in the outermost judgment area. In practical applications, the number of data objects in the result set is often not required. For example, 10 results are returned, so at most you only need to query The data set in the outermost area is fine, so at least 2/3 of the search range is reduced, and a large amount of irrelevant data is filtered out. Therefore, the method of the present invention greatly improves the top-k query performance under massive data, and improves the top-k query speed of massive higher-dimensional data sets.

Description of drawings

Fig. 1 is a schematic diagram of the data division method of the present invention;

Fig. 2 is a schematic diagram of the method for subdividing data in the present invention;

Fig. 3 shows the comparison of three different data division methods with different query time of data dimensions, wherein DistImprove is the method of the present invention, AngleDistTop_k is based on the angle and distance data segmentation method, and BasicTop_k is the query time without dividing the original data set;

Figure 4 shows the comparison of query time for different user input weights in the case of dimension 4.

Fig. 5 is the acceleration ratio of the inventive method at different cluster nodes;

FIG. 6 is a diagram of the top-k query process in the specific implementation of the top-k query in the present invention.

detailed description

The present invention will be further described below in conjunction with the accompanying drawings.

The experiment is done on a 7-node spark cluster. Spark is built on hadoop, using hadoop's yarn resource manager and HDFS file storage system. Among the 7 nodes, the master node is both a driver node and a worker node, and the remaining 6 nodes are all worker nodes. The basic configuration of the experimental environment is shown in Table 2:

Table 2

Using a uniform data set, each record has 8 attributes, and each attribute takes an integer in the value range [0,1000]. A total of 1 billion records are generated, with a data volume of about 40G. At the same time, 4-dimensional and 6-dimensional data sets are also generated, and they are similar to 8-dimensional data sets, which are randomly generated data sets with 1 billion records.

Unless otherwise specified, the following experiments are conducted under the condition of taking the mean value of weights and k=100, and each query is the result of taking the average value 10 times. Since the query data preprocessing only needs to be done once, and then each query does not need to consider the data preprocessing, so the comparison of query time below is not included in the data preprocessing time. The preprocessing time of the method of the present invention is about 42 minutes for 8-dimensional data.

As shown in Figure 6, this embodiment is divided into two major steps:

Step 1: Data preprocessing. Mainly divide the original data set according to the data segmentation method proposed in this paper, divide it into different blocks, mark each block, and then store it on the HDFS disk. In Spark, the HDFS disk is mainly composed of each worker node disk. The query is performed according to the data segmentation method and the query data judgment method in the claims. The specific division is as follows:

The first step: overall segmentation from the inside to the outside

According to the principle of equal area, the entire data space is divided into three large partitions with m=3 from the inside to the outside. As shown in Figure 1, take two-dimensional as an example to illustrate the division method intuitively. The horizontal axis is the x-axis corresponding to attribute 1, and the vertical axis is the y-axis corresponding to attribute 2. The solid line in the figure divides the space into 3 equal parts. Each area is numbered 1, 2, 3 from outside to inside. Except for the outermost area, for each of the remaining areas, the point that belongs to the area and the attribute of each axis is the maximum value is used as the base point, and all the attribute values in the entire coordinate system are greater than or equal to the corresponding attribute value at the base point. The areas are divided and numbered A, B in order from outside to inside. Sequentially judge whether n _A ≥ k, or n _B ≥ k is true. If the square A data volume n _A ≥ k is true, only query the data set in area 1, otherwise check whether the data volume n _A ≥ k in square B established. For massive data queries, in general, the amount of data in square A is much larger than k, and the top-k results can be obtained by only querying the data in area 1. In order to further reduce the amount of query data, each large partition is subdivided.

Step 2: Subdivision of each large area

For each partition 1, 2 is further divided, for example, large partition 1, as shown in Figure 2, can be divided into three areas (ABC), D, and E, wherein the areas of A, B, and C are equal, and A, B, and C are mandatory search areas, and D and E are optional search areas.

There is also the following fact: if for a data object T ₁ score Then it can be known that f _W (T ₁ ) is in a straight line with the spatial data point T ₁ It is proportional to the projection length of , so the scoring function can be measured by the projection length.

Therefore, assuming that the data set in square A is greater than or equal to k, it is only necessary to verify whether D and E need to be retrieved in the following way:

If the weight entered by the user satisfies And w ₁ =w ₂ =0.5, the maximum boundary point d of area D in the comparison figure is on the straight line The distance between the projection point above and the origin and the distance between the base point a of the square A and the projection point of the above-mentioned straight line to the origin can be found to be equal. Similarly, the maximum boundary point e of the E area is on the above-mentioned straight line The distance between the projection point on and the origin is also equal to the distance between the base point a of the square A and the projection point of the above straight line to the origin. Therefore, there is no need to query the D and E areas, only the A, B, and C areas You can get the top-k results, and according to the same reason, you can also know that the top-k results found in the A, B, and C areas will inevitably appear in the upper right corner of the dotted line in Figure 2;

Similar to the above principle, if w ₁ >w ₂ , then there is no need to query area D; if w ₁ <w ₂ , then there is no need to query area E, and we will not discuss them one by one here;

In summary, the following formula can be obtained:

Extended to the d-dimensional data space, use S _i , 1≤i≤3 to represent one of the three large partitions divided from the inside to the outside; S _ij , 1≤j≤d to represent the large partition S _i that is similar to D or E The jth sub-region; S _i(d+1) represents the sub-regions similar to A, B, and C in the large partition S _i . When the weight of the jth attribute of the data object , then only two regions S _i(d+1) and S _ij need to be queried for the large partition S _i ; otherwise, only one region S _i(d+1) needs to be queried for the large partition S _i .

Step 2: Query processing. For a user query f(W,k), select part of the data set to query on the Driver node according to the query input. In this embodiment, k=100 is used, and the weight of each data object attribute is averaged. At this time, only the S _1(d+1) data area is queried.

As shown in Figure 3, three different data division methods are compared with different query time of data dimensions, wherein DistImprove is the method of the present invention, AngleDistTop_k is based on the angle and distance data segmentation method, and BasicTop_k is not to divide the query time for the original data set; from the experiment It can be seen that the method of the present invention improves the query speed more than the data segmentation method based on angle and distance, and the query speed is increased by about 15%, and the query time also increases steadily with the increase of the dimension without large fluctuations.

Since the input of the user's weight W=(w ₁ ,w ₂ ,...,w _d ) will affect the size of the query data area, as shown in Figure 4, the query time of different weight values in the 4 dimensions, where the first category is It has the characteristics of being extremely biased towards a certain attribute weight, including W ₂ =(0.06,0.06,0.07,0.8) and W ₄ =(0.56,0.14,0.25,0.03), the second type W ₁ =(0.25,0.25,0.25, 0.25) is the equivalent weight, and the third category W ₃ =(0.16, 0.32, 0.34, 0.18) is the weight with no obvious bias. In the figure, the query time of W ₁ and W ₃ is approximately equal, the query time of W ₂ and W ₄ is approximately the same, and the query time of W ₁ and W ₃ is shorter than that of W ₂ and W ₄ , mainly due to the different weights in low-dimensional data sets As a result, different data blocks need to be queried, and W ₂ and W ₄ are extremely biased towards a certain attribute weight, resulting in the need to query some data blocks more, resulting in a query time greater than w ₁ and w ₃ .

The scalability of the method of the present invention is shown in Figure 5. The speedup ratio of the 8-dimensional data set on different nodes can be seen to be close to the ideal speedup ratio. As the number of processors, that is, workers doubles, the execution speed will also double. , that is, the data division method of the present invention has good scalability.

The above is only a preferred embodiment of the present invention, it should be pointed out that for those of ordinary skill in the art, without departing from the principle of the present invention, some improvements and modifications can also be made. It should be regarded as the protection scope of the present invention.

Claims (2)

1. A massive data weighted top-k query method based on distributed computing framework, characterized in that: comprise the following steps of sequential execution: Step 1. Create a data space Firstly, the attribute values of all data objects containing d attributes are converted into non-negative values, and the attribute values are normalized; a d-dimensional coordinate system is established, and the axes of the coordinate system correspond to the attributes of the data objects one by one. All data objects are placed in the coordinate system to form a data space; Step 2. Data division Starting from the origin of the coordinate system, divide the entire coordinate system into m areas from the inside to the outside, and number each area from the outside to the inside as 1, 2,..., m, and the boundary of the first area is together with the coordinate axis Include all data objects together. For any region, the maximum value of each attribute of the region is the same, and the coordinates of the outer boundary of each region satisfy at least one axis whose coordinate value is the maximum value of the attribute of the region , on the premise that the maximum value of the attribute of the first region is set to a ₁ , then the maximum value of the attribute of the i-th region Except for the outermost area, for each of the remaining areas, the point that belongs to the area and the attribute of each axis is the maximum value is used as the base point, and all the attribute values in the entire coordinate system are greater than or equal to the corresponding attribute value at the base point. The areas are all divided, numbered 1, 2,..., M in the order from outside to inside, where M=m-1, and the above newly divided areas are used as the judgment area to perform the following judgments: According to the order of numbering from small to large, judge whether N _i ≥ k holds true, where N _i is the number of data objects in the i-judgment area, and k is the number of data objects in the result set returned by the algorithm; when i judges If the area satisfies the above formula and holds, the judgment ends, and the i areas from outside to inside are used as search areas for searching. 2. The mass data weighted top-k query method based on the distributed computing framework according to claim 1, characterized in that: the innermost region in the search region is subdivided, the region number is i, and the subdivision Methods as below: Divide the area into d+1 blocks, where block d is the search area to be selected, and the remaining areas except the search area to be selected are mandatory search areas; Number the search areas to be selected as n=1, 2,...,d, where any data point T _nj (t _n1 ,t _n2 ,...,t _nd ) in the nth search area to be selected, where t _nj Indicates the attribute value corresponding to the j-axis of the data point T _nj , t _nj satisfies the following two inequalities: 0≤t _nj ≤2a _i+1 -a _i , where 1≤j≤d and j≠n (1) a _i -a _i+1 ≤t _nj ≤a _i , where j=n (2) In the nth search area to be selected, if the data point T _nj (t _n1 ,t _n2 ,…,t _nd ) satisfies that the attribute value corresponding to one axis is a _i and the attribute value corresponding to the other axes is 2a _i+1 - a _i , then use this data point as the maximum boundary point of the nth candidate retrieval area; Traversing whether each attribute weight w _j at the maximum boundary point of each candidate retrieval area assigned by the user satisfies the following conditions: If there is an attribute weight w _j of the maximum boundary point that satisfies the above conditions, the scope of the search area is narrowed to include i-1 areas from outside to inside, the required search area in the i-th area, and the candidate to be selected to which the maximum boundary point belongs search area; If there is no attribute weight w _j of the maximum boundary point that satisfies the above conditions, the search area is narrowed to include i-1 areas from outside to inside and the required search area in the i-th area.