CN105930457A - Distributed architecture-based data flow frequent item mining method - Google Patents

Abstract

The invention discloses a distributed architecture-based data flow frequent item mining method. According to the method, a two-layer tree-shaped communication structure which comprises m leaf nodes and 1 root node is adopted, wherein the leaf nodes are responsible for processing data items in data flows and sending frequency increments to the root node when increments of frequencies of the data items exceed a threshold value; the root node is responsible for collecting updating transferred by the leaf nodes. The method is low in communication overhead, and can be used for responding frequent item query requests initiated by users in real time.

Description

Mining method of frequent items in data stream based on distributed architecture

technical field

The invention belongs to the technical field of data mining, and relates to a frequent item mining method, in particular to a frequent item mining method applicable to a distributed architecture.

Background technique

In data mining problems such as association rule mining, sequential pattern mining, correlation mining, and multi-layer pattern mining, frequent item mining is not only a basic step, but also a key step. At present, there are many researches on frequent item mining of data streams on a single node, which can be roughly divided into two categories: counter-based methods and sketch-based methods.

The counter-based frequent item mining method maintains a set of counters to count the frequency of data items. Each counter contains two parameters, the data item name and the data item frequency. When a data item arrives in the data stream, if the data item exists in the maintained counter, the frequency recorded by the counter is increased accordingly; otherwise, a new counter is added to store the new data item or replace an original counter. Usually counter-based frequent item mining methods have low overhead for processing individual data items, but need to sort all counters periodically. Counter-based frequent item mining methods include Frequent, Space Saving, and Lossy Counting.

Sketch-based methods use hash tables composed of one-dimensional or two-dimensional counter arrays to estimate the frequency of individual data items in a data stream. Such methods usually use hash technology to map each data item to corresponding multiple counters, and a single counter may be shared by multiple data items, that is, data items with the same hash value share the same counter. When a data item arrives, only the value of the corresponding counter needs to be modified. When a user submits a query request, the corresponding counter value is used to estimate the frequency, and the error can be controlled within a certain range with a high degree of confidence. In general, hash-based methods require the support of additional data structures, such as using a heap to track candidate frequent items. Skech-based frequent item mining methods include CountSketch, CountMin Sketch and hCount, etc.

Most of the traditional frequent item mining methods only consider processing a single data source on a single node. However, in many practical applications, the data to be mined is large-scale distributed, such as flow detection in network monitoring, DDOS attack detection Wait. At present, the research on frequent item mining methods in the distributed flow environment mainly includes the Tributary-Delta method proposed by A. Manjhi, which is suitable for tree or multi-path graph topology; Approximation methods for frequent items in velocity distributed streams; methods for distributed top-k monitoring proposed by B.Babcock and C.Olston, etc. The main disadvantages of these methods include: communication overhead is too large and real-time query is not supported.

Contents of the invention

The purpose of the present invention is to solve the problems of excessive communication overhead and unsupported real-time query in existing methods, and provide a data stream frequent item mining method based on a distributed architecture, which can be used to improve the traditional frequent item mining method on a distributed architecture. processing power.

The invention provides a method for mining frequent items of data streams based on a distributed architecture, which is an ε-approximation method of a method for mining frequent items of distributed data streams with weights. The method adopts a two-layer tree communication structure, including m leaf nodes and a root node; the leaf nodes are responsible for processing data items in the data stream, and store the frequency of data items in the data stream in the minimum heap of the leaf nodes , and when the data item frequency increment is greater than the threshold, the data item frequency increment is sent to the root node; the root node is responsible for calculating the frequency estimate of the data item in the overall architecture, and storing the data item frequency estimate in the root In the minimum heap of the node; the entries stored in the minimum heap of the leaf node include the data item name, the data item frequency and the data item frequency increment; the entries stored in the root node's minimum heap include the data item name and the data item frequency estimates.

Technical scheme of the present invention:

Step 1), each leaf node i takes out data item successively from received data flow, and described data item comprises data item name v _t and data item frequency c _{v, t} ;

Step 2), updating the sum of data item frequencies of the leaf nodes N _i =N _i +c _v,t and the increment of the sum of data item frequencies Δ _i =Δ _i +c _v,t , where the equal sign expresses assignment, the same below;

Step 3), according to the data item name v _t and the data item frequency c _{v, t} of the data item taken out in step 1), find out a suitable entry in the minimum heap H _i of the leaf node, and for this entry The data item name, data item frequency and data item frequency increment assignment; this step includes:

Step 3-1), judging whether there is an entry whose data item name is v _t in the minimum heap of the leaf node, if there is, execute the next step, otherwise, execute step 3-5);

Step 3-2), judging whether the minimum heap H _i of the leaf node is full, if full, execute the next step, otherwise, execute step 3-4);

Step 3-3), taking out the entry item _min with the smallest data item frequency from the minimum heap H _i of the leaf node, reassigning the entry, and then performing step 4); wherein, assigning the entry includes:

Let v=vt, _cv = _cv + _{cv,t , Δv} = _cv,t ;

The v represents the data item name of the fetched entry, the v _t represents the data item name of the fetched data item, the c _v represents the data item frequency of the fetched entry, and the c _{v, t} represents the fetched The data item frequency of the data item, the _Δv represents the frequency increment of the data item of the fetched item;

Step 3-4), creating a new entry and assigning a value to the new entry, inserting the new entry into the minimum heap H _i of the leaf node, and then performing step 4); wherein, assigning a value to the new entry includes:

Let v=vt, _cv = _{cv,t , Δv} = _cv,t ;

Step 3-5), take out the existing entry from the minimum heap H _i of the leaf node And update the entry, and then perform step 4); wherein, updating the entry includes:

Let _cv = _cv + _cv,t , _Δv = _Δv + _cv,t ;

Step 4), judging whether the increment of the sum of the data item frequencies and the data item frequency increment of the entry is greater than a threshold, if greater than the threshold, transfer an update to the root node; this step includes:

Step 4-1), judging whether the increment Δ _i of the sum of data item frequencies satisfies Δ _i > β _i N _i , if so, execute the next step, otherwise, execute step 4-3); wherein,

The β _i represents the update delay coefficient of the leaf node defined by the user, and the N _i represents the sum of the data item frequencies of the leaf node;

Step 4-2), the leaf node sends a 0-msg update to the root node, and then sets the value of _Δi to 0; wherein

The content sent by the 0-msg update includes the increment _Δi of the frequency sum of the data items;

Step 4-3), whether the data item frequency increment Δ _v of the entry satisfies Δ _v > β _i N _i , if so, execute the next step, otherwise, execute step 5);

The β _i represents the update delay coefficient of the leaf node defined by the user, and the N _i represents the sum of the data item frequencies of the leaf node;

Step 4-4), the leaf node sends a data item update to the root node, and then sets the value of _Δv to 0; wherein

The content sent by the data item update includes the data item name of the entry and the data item frequency increment _Δv of the entry;

Step 5), the root node sequentially fetches updates from the updates sent by the leaf nodes, and maintains corresponding data according to the fetched updates; this step includes:

Step 5-1), determine the type of update sent by the leaf node taken out by the root node, if it is a 0-msg update, perform the next step, if it is a data item update, perform step 5-3);

Step 5-2), updating the estimated value N=N+Δ _i of the sum of the data item frequencies of the root node, wherein the equal sign indicates assignment, and then perform step 6); wherein,

The N represents the estimated value of the sum of the data item frequencies of the root node, and the _Δi represents the frequency of the 0-msg update sent by the leaf node;

Step 5-3), updating the minimum heap H ₀ of the root node; this step includes:

Step 5-3-1), taking out the data item name v _t and data item frequency increment Δ _{v, t} in the update sent by the leaf node;

Step 5-3-2), judging whether there is an entry item _v whose data item name is v _t in the minimum heap of the root node, if there is, execute the next step, otherwise, execute step 5-3-4);

Step 5-3-3), take out the entry item _v , update and update the entry, and then perform step 6); wherein, updating the entry includes:

Let c _v =c _v +Δ _v,t , where the equal sign indicates assignment;

The v represents the data item name of the entry, the c _v represents the data item frequency of the entry, and the Δv _{, t} represents the frequency increment of the data item for the update of the data item;

Step _5-3-4 ), judging whether the minimum heap H0 of the root node is full, if full, execute the next step, otherwise execute 5-3-6);

Step 5-3-5), taking out the entry item _min with the smallest data item frequency in the minimum heap _H0 of the root node, reassigning the entry, and then performing step 6); wherein, assigning the entry includes:

Let v=vt, _cv = _cv +Δv _{,t ;}

The v _t represents the data item name of the updated data item taken out;

Step 5-3-6), creating a new entry and assigning a value to the new entry, inserting the new entry into the minimum heap maintained by the root node, and then performing step 6); wherein, assigning a value to the new entry includes:

Let v=vt, _cv =Δv _{,t ;}

Step 6), according to the user's request, the root node traverses the minimum heap H ₀ , and returns the frequency of all data items The entries of are the frequent items to be mined.

The present invention also includes an operation step of sorting the minimum heap H _i of the leaf nodes according to the frequency of data items between step 4) and step 5).

And, an operation step of sorting the minimum heap H0 of the root node according to the frequency of the data items is also included between the steps ₅ ) and 6).

Advantages and beneficial effects of the present invention:

The error between the estimated value of the data item frequency of the frequent item output by the method provided by the present invention and the real value is not greater than εN, and the maximum communication overhead on a single link is not greater than It can support users to query frequent items in real time.

Description of drawings

Figure 1 is the communication structure of the method for mining frequent items in data streams based on distributed architecture.

Figure 2 is the average relative error of the data stream frequent item mining method based on the distributed architecture.

Figure 3 shows the single-link communication overhead of the data stream frequent item mining method based on the distributed architecture.

Figure 4 is the initialization time of the data stream frequent item mining method based on the distributed architecture.

detailed description

In order to express the method idea of the present invention more clearly and intuitively, the details of the method for mining frequent items of data streams based on the distributed architecture are described in detail below:

1. Determine the parameters

The parameters that need to be determined in the distributed frequent item mining method include:

(1) Data item support and degree of error

(2) Leaf node data m;

(3) The size of the minimum heap H _i and H ₀ of each leaf node i and root node is and

(4) Delay update coefficient β _i (0<β _i <ε) of each leaf node i.

In this embodiment, the number of leaf nodes is m=8, and the minimum heap sizes of leaf nodes and root nodes are both 10000, that is, α ₀ =α _i =0.0001, β _i ∈[0.001,0.005], ε∈[0.0001,0.0005] ],

2. Initialization

Determine the initialization time of each leaf node i as B _i >0 is the bandwidth of the link between the leaf node i and the root node, and the unit is data packet/second. At the initial moment of the operation of the method of the present invention, each leaf node i will process the data item update received from the data flow S _i , but will not pass any message to its root node until the initialization state ends.

3. Leaf nodes process data items in the data stream

When a new data item (v, c _{v, t} ) in the data stream S _i arrives at the leaf node i, first update the sum of the frequency of the data item in the data stream S _i N _i =N _i +c _{v, t} , where the equal sign Indicates the assignment, the same below, and the frequency estimation value increment Δ _i =Δ _i +c _v,t of the sum of the frequencies of the data items. Then update the corresponding data item frequency: if v∈H _i , then increase the frequency estimate value of the corresponding data item entry _cv = _cv + _{cv, t} and the frequency estimate value increment of data item _v Δv = _Δv + c _{v, t} ; otherwise, find the data item _min with the smallest frequency estimate in H _i , replace the data item name of item _min with v, and update its data item frequency c _v = c _v + c _{v, t} and frequency estimate Value increment Δ _v =c _v,t . Finally, check whether the condition is satisfied and transmit the data item update to the root node: if Δ _i > β _i N _i , then send the update (0, Δ _i ) to the root node, and reset the frequency estimation increment Δ _i = 0; if Δ _v >β _i N _i , then send an update (v, Δ _v ) to the root node, and reset the increment of the frequency estimation value Δ _v =0.

4. The root node processes the updates sent by the leaf nodes

When the root node receives the data item update (v, Δ _v ) delivered by the leaf node, if v∈H ₀ , then increase the frequency estimation value of the corresponding data item entry c _{v =} c _v +Δ _v ; otherwise, find For the data item item _min with the smallest estimated frequency value, replace the data item name of item _min with v, and update its data item frequency c _v =c _v +Δ _v . When the root node receives the 0-msg update (0, Δ _i ) delivered by the leaf node, it updates the root node's estimated value N ₀ =N ₀ +Δ _i of the sum of data item frequencies.

5. The root node handles frequent item queries initiated by users

When a user submits a frequent item query to the root node, the root node scans each data item entry (v, c _v ) maintained in the min-heap H ₀ . If cv _≥ ωN ₀ , v is considered to be a frequent item and v will be output, where ω is the output threshold, there is is the user-defined data item support, α _max =max(α _i ), β _max =max(β _i ).

The present invention is implemented by means of real data and computer simulation.

The present invention selects 3 groups of network traffic data sets collected in a real network environment as data sources in the embodiment. These three sets of data sets are: CERNET data set, which is a TCP bidirectional data set collected on the OC-48 link of CERNET (China Education and Research Network); CAIDA48 data set, which is collected on the OC-48west coast peering link The anonymous data set of CAIDA192 data set is a single anonymous data set collected on the OC-192 link. The present invention defines the quintuple (source IP address, destination IP address, source port, destination port, transport layer protocol) of the IP data packet in the network flow data set as the data item name, and defines the length of the data packet load as the data item frequency .

Define the relative value of the update delay coefficient β as Fig. 2 shows the average relative error of the method of the present invention for processing 3 different network datasets. It can be observed that when ε∈[0.0001, 0.0005], the average relative error of the method is smaller than the product of the current ε value and N.

Fig. 3 shows the single-link overhead of the method of the present invention for processing three different sets of network data sets. For each sub-graph in Figure 3, there are 4 curves, and each curve from top to bottom represents the theoretical maximum value of the single-link communication overhead for processing the current network data set Actual maximum, actual average, and actual minimum of single-link communication overhead. It can be observed that the actual maximum communication overhead on a single link is not greater than

Figure 4 shows the initialization time of the method of the present invention. It can be observed that the larger the relative value of the update delay coefficient β, the less the initialization time required by the method of the present invention.

Claims (3)

1. A data flow frequent item mining method based on a distributed architecture adopts a 2-layer tree-shaped communication structure, and comprises m leaf nodes and 1 root node; the leaf node is responsible for processing data items in the data stream, storing the frequency of the data items in the data stream in a minimum heap of the leaf node, and sending the frequency increment of the data items to the root node when the frequency increment of the data items is greater than a threshold value; the root node is responsible for calculating the frequency estimation value of the data item in the whole framework and storing the frequency estimation value of the data item in the minimum heap of the root node; the entries stored in the minimal heap of leaf nodes include data item names, data item frequencies, and data item frequency increments; the entries stored in the minimum heap of the root node include data item names and data item frequency estimates; the method comprises the following steps:

step 1), each leaf node i sequentially takes out data items from the received data stream, wherein the data items comprise data item names v_tAnd frequency c of data items_v，t；

Step 2), updating the sum N of the data item frequencies of the leaf nodes_i＝N_i+c_v，tAnd the increment delta of the sum of the frequencies of the data items_i＝Δ_i+c_v，tWherein the equal sign represents the assignment, the same as follows;

step 3) of retrieving the data item name v of the data item according to step 1)_tAnd data item frequency c_v，tMinimum heap H at said leaf node_iFinding out a proper entry, and assigning values to the name of the data item, the frequency of the data item and the frequency increment of the data item in the entry; the method comprises the following steps:

step 3-1), judging whether the data item name v exists in the minimum heap of the leaf nodes_tIf the item (3) exists, executing the next step, otherwise, executing the step 3-5);

step 3-2), judging the minimum pile H of the leaf nodes_iWhether the container is full, if so, executing the next step, otherwise, executing the step 3-4);

step 3-3), from said minimum heap of leaf nodes H_iItem with minimum data item frequency_minReassigns the entry and then performs step 4); wherein assigning the entry comprises:

let v equal v_t，c_v＝c_v+c_v，t，Δ_v＝c_v，t；

Said v representing the name of the data item from which the entry was fetched, said v_tA data item name indicating a retrieved data item, said c_vFrequency of data items representing fetched entries, said c_v，tData item frequency, said Δ, representing a retrieved data item_vA data item frequency increment representing a fetched entry;

step 3-4), creating a new entry and assigning a value to the new entry, and inserting the new entry into the minimum heap H of the leaf nodes_iThen step 4) is performed; wherein assigning the new entry comprises:

let v equal v_t，c_v＝c_v，t，Δ_v＝c_v，t；

Step 3-5), from said minimum heap of leaf nodes H_iFetching an existing entryUpdating the entry, and then executing the step 4); wherein updating the entry comprises:

let c_v＝c_v+c_v，t，Δ_v＝Δ_v+c_v，t；

Step 4), judging whether the increment of the sum of the data item frequencies and the data item frequency increment of the entry are greater than a threshold value, and if so, transmitting the update to the root node; the method comprises the following steps:

step 4-1), judging increment delta of the sum of the data item frequencies_iWhether or not Δ is satisfied_i＞β_iN_iIf yes, executing the next step, otherwise, executing the step 4-3); wherein,

β as described_iAn update delay factor representing a user-defined leaf node, said N_iA sum of data item frequencies representing the leaf nodes;

step 4-2), the leaf node sends 0-msg update to the root node, and then the delta is updated_iIs set to 0; wherein

Said 0-msg update transmitted content comprises an increment delta of the sum of said data item frequencies_i；

Step 4-3), data item frequency increment delta of said item_vWhether or not Δ is satisfied_v＞β_iN_iIf yes, executing the next step, otherwise, executing the step 5);

β as described_iAn update delay factor representing a user-defined leaf node, said N_iA sum of data item frequencies representing the leaf nodes;

step 4-4), the leaf node sends data item update to the root node, and then the delta is sent_vIs set to 0; wherein

The data item updating transmission content comprises the data item name of the item and the data item frequency increment delta of the item_v；

Step 5), the root node sequentially takes out updates from the updates sent by the leaf nodes and maintains corresponding data according to the taken-out updates; the method comprises the following steps:

step 5-1), judging the type of the update sent by the leaf node taken out by the root node, if the type is 0-msg update, executing the next step, and if the type is data item update, executing step 5-3);

step 5-2), updating the estimated value N of the sum of the data item frequencies of the root node to be N + delta_iWherein the equal sign represents the assignment, and then step 6) is performed; wherein,

said N representing an estimate of the sum of the data item frequencies of the root node, said Δ_iIndicating a frequency of 0-msg updates sent by said leaf node;

step 5-3), updating the minimum heap H of the root node₀(ii) a The method comprises the following steps:

step 5-3-1), the data item name v in the update sent by the leaf node is taken out_tAnd data item frequency increment delta_v，t；

Step 5-3-2), judging whether the data item name v exists in the minimum heap of the root node_tItem of_vIf yes, executing the next step, otherwise, executing the step 5-3-4);

step 5-3-3), the item is taken out_vUpdating the entry, and then executing step 6); wherein updating the entry comprises:

let c_v＝c_v+Δ_v，tWherein the equal sign represents the assignment;

saidv denotes the data item name of the fetched entry, said c_vFrequency of data items representing fetched entries, said Δ_v，tA data item frequency increment representing a fetch data item update;

step 5-3-4), judging the minimum heap H of the root node₀Whether the container is full, if so, executing the next step, otherwise, executing 5-3-6);

step 5-3-5), taking out the minimum heap H of the root node₀Item with minimum frequency of medium data item_minReassigns the entry and then performs step 6); wherein assigning the entry comprises:

let v equal v_t，c_v＝c_v+Δ_v，t；

V is_tA data item name indicating a retrieved data item update;

step 5-3-6), creating a new entry and assigning a value to the new entry, inserting the new entry into the minimum heap maintained by the root node, and then executing step 6); wherein assigning the new entry comprises:

let v equal v_t，c_v＝Δ_v，t；

Step 6), according to the request of the user, the root node traverses the minimum heap H₀Frequency of all data items returnedThe entries of (1) are frequent items to be mined.

2. The method for mining frequent items of data flow based on distributed architecture as claimed in claim 1, further comprising a minimum heap H of leaf nodes between said steps 4) and 5)_iThe operation steps of sorting by frequency of the data items are performed.

3. The method for mining frequent items of data flow based on distributed architecture as claimed in claim 1, further comprising a minimum heap of root nodes between said steps 5) and 6)H₀The operation steps of sorting by frequency of the data items are performed.