CN109726225A - A Storm-based Distributed Streaming Data Storage and Query Method - Google Patents

Abstract

The present invention provides a method for storing and querying distributed stream data based on Storm. The present invention is based on the Storm data stream computing framework, and CEPHFS is used as the data underlying storage system to perform real-time data analysis by analyzing the characteristics of distributed stream data. The partition and index construction are performed, and the partitioned data blocks are compressed and stored in CEPHFS. During the search operation, according to the attributes of the key and temporal dimensions of the data block, the query is decomposed into corresponding sub-queries, and only the files that may contain the required data are read through the bloomFilter method, and the qualified data is selected by the predicate and submitted. After the sub-query results are merged, the aggregate operation is performed and returned to the user. Make full use of computing resources to improve the efficiency of data storage and query. The invention has the characteristics of wide application scenarios, low time delay, load balance, and can realize high-speed storage.

Description

A Storm-based Distributed Streaming Data Storage and Query Method

technical field

The invention relates to the technical field of data processing, in particular to a method for storing and querying distributed stream data based on Storm.

Background technique

With the rapid development of network technology and the rapid growth of real-time streaming data generated by social networks and location-based service platforms, the requirement for real-time processing and response to massive streaming data has emerged in more and more fields. Real-time search has become a very important data processing capability, and users can obtain desired historical data and new data in real time. For platforms that provide location services, such as Baidu Maps, AutoNavi Maps, etc., massive amounts of location information and trajectory change data are instantaneously generated every second. In order to meet the needs of users and improve company benefits, the platform system needs to be able to support millions of streams. Real-time insertion storage and low-latency query on data. For example, customers need to obtain GPS information of all vehicles within 5km of the current moment, or specify the driving trajectory of a vehicle in the past 1 hour.

Common open-source implementations of key-value storage technologies, such as HBase, use LSM-Tree to reduce the time overhead caused by updating leaf nodes, but new data and historical data inserted each time need to be updated, and the query delay in the time range is too high; Commonly used time series database technologies such as Alibaba's open-source Druid only support inverted indexes, which are inefficient for key range queries. In order to solve this problem, it is necessary to design a distributed database technology that can perform high-speed storage and real-time query for massive stream data. Both key range and time range support efficient query, which requires new data and historical data to have Separated, try to avoid the traversal of irrelevant data when querying, improve query efficiency, and ensure the load balance of different nodes in the system to maximize resource utilization.

SUMMARY OF THE INVENTION

In view of the deficiencies of the prior art, the present invention provides a method for storing and querying distributed stream data based on Storm. The present invention analyzes the stability characteristics that stream data will arrive in close order and data distribution range in a real environment, and the current database The technology cannot meet the high-efficiency problem of key range and time range, and provides a real-time query processing method for efficient indexing and time domain range under massive flow data. The purpose of the invention is to divide the range of the upcoming stream data, store them in the distributed file system after parallel indexing on different machine nodes, perform query decomposition during query, execute sub-query, filter, aggregate and other operations in parallel, and merge the results. return.

The technical scheme of the present invention is as follows: a method for storing and querying distributed stream data based on Storm, the present invention establishes several B+Tree indexes with isolated ranges in real time when receiving distributed stream data, and stores them in distributed files after reaching a threshold value. system, and perform query decomposition when querying, process sub-queries in different scopes in parallel, maintain load balance, merge and return real-time storage results after completion, and realize high-throughput streaming data insertion and query, including the following steps:

S1), receive source data and distribute to downstream units to build an index structure;

S2), compress the index structure into data blocks and write into the distributed file storage system CEPHFS;

S3), decompose the query into several independent sub-queries based on query conditions and data block information;

S4), distribute to the sub-query of the downstream independent query processing unit by accessing the distributed file storage system CEPHFS;

S5), receive the returned sub-query results and combine them and return them to the user.

Further, in step S1), each source data received by the streaming data storage system is a data tuple, which is defined as d={d _k , d _t , d _r }, where d _k is the primary key of the tuple, and d _t is the time Attribute, d _r is other attribute values of the tuple, K and T define a two-dimensional space of primary key and time domain D=(K, T); the primary key range is fixed, the time range is increasing, the primary key K interval is expressed as K(k -,k+), the time domain T interval is expressed as T(t-,t+), and a unique rectangle r≤K is established according to the two intervals, T≥{(k,t)∈R|k∈K,t∈T} .

Further, write the data tuples within the range of the rectangle r≤K, T≥{(k,t)∈R|k∈K,t∈T} into the unique corresponding template B+Tree, the key is used as the index, the memory The template B+Tree that reaches the threshold chunkSize size is stored in the distributed file system in the form of chunks. The chunk consists of a key array and a data array. The key array stores the sequential key values, including an offset to the data array.

Further, based on the two-dimensional space D=(K, T), the query condition of the streaming data storage system can be defined as a triple q={K _q , T _q , f _q }, where K _q , T _q are in the primary key and the conditional selection range in the time domain, the query interval is divided into a r≤K, T≥{(k,t)∈R|k∈K _q ,t∈T _q },f _q :t->{true, false} is a user-defined conditional filter function to determine whether the user's choice is satisfied.

Further, based on the different file blocks stored by different subquery server Subquery Server nodes and the different cached template B+Tree leaf nodes, the algorithm of query decomposition and scheduling is implemented, and the priority queue of each unprocessed subquery by the subquery server SubqueryServer is calculated. Query allocation is performed until the set of unprocessed subqueries is empty, and the recently queried leaf node data is written into the cache to achieve cache locality, data block locality and load balancing of query allocation. The specific algorithm process is as follows:

For S(q _i ) and Shuffle, if S(q _i ) comes first, splicing the two into a new array Among them, the smaller the subscript represents the higher priority, the The element containing the priority is added to the subquery priority queue of each subquery server Subquery Server, after all _qi are processed, the priority queue of the subquery server Subquery Server is taken out in turn with the highest priority and unprocessed. q _i is allocated until all q _i are allocated, where S(q _i ) _represents the Subquery Server array that stores the qi range data, Represents the array of other subquery servers Subquery Server, q _i ∈ q represents the subquery after a query is decomposed.

Further, in step S2), the index structure is a tree-shaped index structure, and after the size of the tree-shaped index structure exceeds a specified threshold, the data element ancestors in the leaf nodes are compressed by the Snappy algorithm, and are written into the distribution in the form of data blocks. It is permanently stored in CEPHFS, a file storage system, and records the metadata of the data block's ancestral primary key and the metadata related to the time domain to the metadata manager metadata keeper; The domain will continue to grow, and the non-leaf node part of the tree index structure is reserved as a template, so that the index template can be used directly in the next construction, avoiding the splitting of nodes like building a B+ tree, causing a lot of time overhead .

Further, in step S3), the query is decomposed into several independent sub-queries based on the query conditions and the data block information, which specifically includes the following steps:

S301), the query dispatcher, according to the primary key and the time domain range in the query condition provided by the user, reads the metadata information of the data block in the metadata manager (metadata keeper) for comparison, and divides the query area into a series of two-dimensional the index area;

S302), based on the equivalent judgment condition provided by the user, filter out the sub-query area that must not contain the target data element ancestor by the bloom filter bloomFilter method;

S303), only distribute the subquery that may contain the target data element ancestor to the downstream independent subquery server Subquery Server.

Further, in step S4), the sub-query distributed to the downstream independent query processing unit by accessing the distributed file storage system CEPHFS specifically includes the following steps:

S401), the subquery server Subquery Server reads the data block corresponding to the subquery in the distributed file storage system CephFs in parallel, first reads the template part of the index structure in the data block, and obtains the relative offset and grouping of the leaf node to all leaf nodes The compressed offset is calculated to obtain a series of leaf node offsets that may include the target key range;

S402), read the leaf node part of the index structure in the data block file based on the offset, decompress the obtained leaf node group data block bytes through the Snappy algorithm, deserialize it into a leaf node, and filter on the time range and equivalent conditions ;

S403), perform an aggregate operation on the filtered series of data primitives, serialize them, and send them to the query dispatcher.

The beneficial effects of the present invention are:

1. The present invention has a wide range of application scenarios, such as distributed stream data processing applications such as communication companies monitoring and analyzing network traffic, location networking platform vehicle traffic trajectory changes, e-commerce platform holiday real-time transaction indicators, etc., to achieve real-time data transmission and processing of massive data.

2. The present invention can realize high-speed storage. The present invention uses an efficient data division method to separate new data and historical data, and uses the characteristics of data range stability to build a B+Tree index by retaining an index template to avoid tree node splitting. A lot of time consuming to come.

3. The present invention has the characteristics of low latency. After the range of query conditions is segmented, only files whose meta-information may meet the query range are accessed, and key operations such as filtering and aggregation are processed in parallel, and cache locality and file locality are realized. Improve query efficiency.

4. The load balancing of the present invention distributes the decomposed sub-queries to different nodes through the designed query scheduling algorithm, and makes full use of system resources.

Description of drawings

Fig. 1 is the schematic flow chart of the present invention;

Fig. 2 is the structure diagram of distributed stream data storage data block of the present invention;

Fig. 3 is the internal structure diagram of the leaf node in the distributed stream data storage data block of the present invention;

FIG. 4 is a diagram of decomposition and scheduling of distributed stream data query according to the present invention.

Detailed ways

The specific embodiments of the present invention will be further described below in conjunction with the accompanying drawings:

As shown in Figure 1 , a method for storing and querying distributed stream data based on Storm, the present invention establishes several B+Tree indexes with isolated ranges in real time when receiving distributed stream data, and stores them in a distributed file system after reaching a threshold value. , and perform query decomposition when querying, process sub-queries under different scopes in parallel, maintain load balance, merge and return real-time storage results after completion, and realize high-throughput streaming data insertion and query. The specific steps include the following steps:

S1), receive source data and distribute to downstream units to build an index structure;

Among them, each source data received by the streaming data storage system is called a data tuple, and is defined as d={d _k , d _t , d _r }, where d _k is the primary key of the tuple, d _t is the time attribute, and d _r are other attribute values of the tuple, K and T define a two-dimensional space of primary key and time domain D=(K, T); the primary key range is fixed, the time range is increasing, the primary key K interval is expressed as K(k-,k+) , the time domain T interval is represented as T(t-,t+), and a unique rectangle is established according to the two intervals:

r≤K, T≥{(k,t)∈R|k∈K,t∈T};

Write the data tuple within the range of the rectangle r≤K,T≥{(k,t)∈R|k∈K,t∈T} into the unique corresponding template B+Tree, the key is used as the index, when the memory reaches The template B+Tree with the threshold chunkSize size is stored in the distributed file system in the form of data chunks. The chunk consists of a key array and a data array. The key array stores the sequential key values, including an offset to the data array.

Based on the two-dimensional space D=(K, T), the query condition of the streaming data storage system can be defined as a triple q={K _q , T _q , f _q }, where K _q , T _q are in the primary key and time domain On the conditional selection range, the query interval is divided into a r≤K,T≥{(k,t)∈R|k∈K _q ,t∈T _q },f _q :t->{true,false} is User-defined conditional filter function to determine whether the user's choice is satisfied.

Based on the different file blocks stored in the Subquery Server nodes of different subquery servers and the difference in the cached template B+Tree leaf nodes, the query decomposition and scheduling algorithm is implemented, and the subquery server Subquery Server is calculated to query each unprocessed subquery priority queue. Allocate until the set of unprocessed subqueries is empty, and write the recently queried leaf node data into the cache to achieve cache locality, data block locality and load balancing of query allocation; the specific algorithm process is as follows:

For S(q _i ) and Shuffle, if S(q _i ) comes first, splicing the two into a new array Among them, the smaller the subscript represents the higher priority, the The element containing the priority is added to the subquery priority queue of each subquery server Subquery Server, after all _qi are processed, the priority queue of the subquery server Subquery Server is taken out in turn, the highest priority and unprocessed ones are taken out. q _i is allocated until all q _i are allocated, where S(q _i ) _represents an array of Subquery Servers (Subquery Servers) that store qi range data, Represents the array of other subquery servers Subquery Server, q _i ∈ q represents the subquery after a query is decomposed.

S2), compress the index structure into data blocks and write into the distributed file storage system CEPHFS; wherein,

The index structure is a tree index structure. After the size of the tree index structure exceeds the specified threshold, the data element ancestors in the leaf nodes are compressed by the Snappy algorithm, and written in the form of data blocks to the distributed file storage system CEPHFS for permanent storage. , and record the metadata related to the primary key of the data block and the time domain to the metadata manager metadata keeper; according to the stream data, the key primary key domain will change within a certain range, and the time domain will continue to grow. The non-leaf node part of the tree index structure is reserved as a template, so that the index template can be directly used in the next construction, and the node splitting is avoided like building a B+ tree, which causes a lot of time overhead.

S3), decompose the query into several independent sub-queries based on the query conditions and data block information, which specifically includes the following steps:

S301), the query dispatcher, according to the primary key and the time domain range in the query conditions provided by the user, reads the data block metadata information in the metadata manager metadata keeper for comparison, and divides the query area into a series of two-dimensional indexes area;

S302), based on the equivalent judgment condition provided by the user, filter out the sub-query area that must not contain the target data element ancestor by the bloom filter method;

S303), only distribute the subquery that may contain the target data element ancestor to the downstream independent subquery server Subquery Server.

S4), distribute the sub-query distributed to the downstream independent query processing unit by accessing the distributed file storage system CEPHFS, which specifically includes the following steps:

S401), the subquery server Subquery Server reads the data block corresponding to the subquery in the distributed file storage system CephFs in parallel, first reads the template part of the index structure in the data block, and obtains the relative offset and grouping of the leaf node to all leaf nodes The compressed offset is calculated to obtain a series of leaf node offsets that may include the target key range;

S402), read the leaf node part of the index structure in the data block file based on the offset, decompress the obtained leaf node group data block bytes through the Snappy algorithm, deserialize it into a leaf node, and filter on the time range and equivalent conditions ;

S403), perform an aggregate operation on the filtered series of data primitives, serialize them, and send them to the query dispatcher.

S5), receive the returned sub-query results and combine them and return them to the user.

As shown in Figure 2, the stream data is written to the internal structure of the chunk file of the distributed file system. The chunk contains the B+Tree template part and the leaf node part. The template in the figure represents the B+Tree template part, the leaf node represents the leaf node part, and the compress chunk represents the data block grouped and compressed by the leaf node.

The B+Tree template part includes the root node and the internal node part of the B+Tree. Each node records the key value, child nodes, etc. The highest internal node also records the relative offset of a list of leaf nodes in all leaf nodes. , the offset of a list of leaf nodes in the chunk after grouping compression.

Leaf nodes include the key array part and the data array part, and all nodes are stored consecutively from left to right. When storing files, the template part is written into the chunk as a whole, and the leaf nodes are written into the chunk in the form of group compression. The number of leaf nodes in each group is set to 20, and the compression ratio is increased to deal with the problem of space storage.

As shown in Figure 3, the data layout of the leaf nodes in the stream data storage structure chunk. The data layout consists of two parts, one is the key array and the other is the data array. The index array in the figure represents the key array, and the data array represents the data array. The key value in the storage order of the Key array includes an offset pointing to the data array. When searching, the key value and offset that meet the conditions are found in the Key array, and then the corresponding data tuple is obtained from the data array.

As shown in Figure 4, the algorithm for handling query decomposition scheduling can be represented as a graph. The Pending Set in the figure represents all the subqueries that have not been assigned, and S(q _i ) represents the optimal assigned Subquery Server for each subquery, Subquery priority array representing each subquery, PreferredServer Arrays represents the priority queue of the Subquery Server for all unprocessed subqueries. When the Pending Set is not empty, each Subquery Server prioritizes the subqueries in the Set according to the local data range stored in the file system and the data range in the cache. The subqueries that have not been processed in the PreferredServer Arrays of the preferred server queue are allocated until all subqueries of the Pending Set are processed.

What is described in the above-mentioned embodiments and specification is only to illustrate the principle and best embodiment of the present invention. Without departing from the spirit and scope of the present invention, the present invention will also have various changes and improvements, and these changes and improvements all fall within the scope of the present invention. within the scope of the claimed invention.

Claims (8)

1. a distributed stream data storage and query method based on Storm, it is characterized in that: by setting up the B+Tree index of some isolation scopes in real time when receiving distributed stream data, after reaching the threshold, store to the distributed file system, and Query decomposition is performed during query, parallel processing of sub-queries in different scopes, load balance is maintained, and after completion, the results are merged and returned to real-time storage to achieve high-throughput streaming data insertion and query, including the following steps: S1), receive source data and distribute to downstream units to build an index structure; S2), compress the index structure into data blocks and write into the distributed file storage system CEPHFS; S3), decompose the query into several independent sub-queries based on query conditions and data block information; S4), distribute to the sub-query of the downstream independent query processing unit by accessing the distributed file storage system CEPHFS; S5), receive the returned sub-query results and combine them and return them to the user. 2. a kind of distributed stream data storage and query method based on Storm according to claim 1, is characterized in that: in step S1), each source data that stream data storage system receives is data element ancestor, is defined as d= {d _k ,d _t ,d _r }, where d _k is the primary key of the tuple, d _t is the time attribute, d _r is the value of other attributes of the tuple, K and T define a two-dimensional space D of the primary key and time domain =(K,T); the primary key range is fixed, and the time range is continuously increasing. The primary key K interval is represented as K(k-,k+), and the time domain T interval is represented as T(t-,t+). According to the two intervals, a unique Rectangle r≤K, T≥{(k,t)∈R|k∈K,t∈T}. 3. A Storm-based distributed stream data storage and query method according to claim 2, characterized in that: the rectangle r≤K, T≥{(k,t)∈R|k∈K,t∈ The data tuples in the range of T} are written into the unique corresponding template B+Tree, the key is used as the index, the template B+Tree that reaches the threshold chunkSize size in the memory is stored in the distributed file system in the form of chunks, and the chunks are composed of key arrays and data. Array composition. The key array stores the sequential key values, including an offset to the data array. 4. a kind of distributed stream data storage and query method based on Storm according to claim 3, is characterized in that: based on two-dimensional space D=(K, T), the query condition of stream data storage system can be defined as a Triple q={K _q , T _q , f _q }, K _q , T _q are the conditional selection range in the primary key and time domain, and the query interval is divided into a r≤K, T≥{(k,t )∈R|k∈K _q ,t∈T _q },f _q :t->{true,false} is a user-defined conditional filtering function, which is used to judge whether the user's choice is satisfied. 5. a kind of distributed stream data storage and query method based on Storm according to claim 4, is characterized in that: the template B+Tree leaf that the file block stored based on different subquery server (Subquery Server) nodes is different and cached Different nodes, implement the algorithm of query decomposition and scheduling, and calculate that the subquery server Subquery Server performs query allocation on each unprocessed subquery priority queue until the unprocessed subquery set is empty and writes the latest query leaf node data Cache, realize cache locality, data block locality and load balancing of query allocation; the specific algorithm process is as follows: For S(q _i ) and Shuffle, if S(q _i ) comes first, splicing the two into a new array Among them, the smaller the subscript represents the higher priority, the The element containing the priority is added to the subquery priority queue of each subquery server Subquery Server, after all _qi are processed, the priority queue of the subquery server Subquery Server is taken out in turn, the highest priority and unprocessed ones are taken out. q _i is allocated until all q _i are allocated, where S(q _i ) _represents the Subquery Server array that stores the qi range data, Represents the array of other subquery servers Subquery Server, q _i ∈ q represents the subquery after a query is decomposed. 6. a kind of distributed stream data storage and query method based on Storm according to claim 1, is characterized in that: in step S2), index structure is tree-shaped index structure, and tree-shaped index structure size exceeds specified threshold value After that, the data ancestry in the leaf node is compressed by the Snappy algorithm, written in the form of data blocks to the distributed file storage system CEPHFS for permanent storage, and the ancestral key of the data block and the metadata related to the time domain range are recorded to Metadata manager metadata keeper. 7. a kind of distributed stream data storage and query method based on Storm according to claim 1, is characterized in that: in step S3), based on query condition and data block information, query is decomposed into some independent sub-queries, concrete Include the following steps: S301), the query dispatcher, according to the primary key and the time domain range in the query conditions provided by the user, reads the data block metadata information in the metadata manager metadata keeper for comparison, and divides the query area into a series of two-dimensional indexes area; S302), based on the equivalent judgment condition provided by the user, filter out the sub-query area that must not contain the target data element ancestor by the bloom filter bloomFilter method; S303), only distribute the subquery that may contain the target data element ancestor to the downstream independent subquery server Subquery Server. 8. a kind of distributed stream data storage and query method based on Storm according to claim 1, is characterized in that: in step S4), distribute to the child of downstream independent query processing unit by accessing distributed file storage system CEPHFS The query includes the following steps: S401), the subquery server Subquery Server reads the data block corresponding to the subquery in the distributed file storage system CephFs in parallel, first reads the template part of the index structure in the data block, and obtains the relative offset and grouping of the leaf node to all leaf nodes The compressed offset is calculated to obtain a series of leaf node offsets that may include the target key range; S402), read the leaf node part of the index structure in the data block file based on the offset, decompress the obtained leaf node group data block bytes through the Snappy algorithm, deserialize it into a leaf node, and filter on the time range and equivalent conditions ; S403), perform an aggregate operation on the filtered series of data primitives, serialize them, and send them to a query dispatcher (query dispatcher).