CN102779185B - High-availability distribution type full-text index method - Google Patents

Abstract

The invention discloses a high-availability distribution type full-text index method. The method comprises the following steps of: firstly, starting a basic service system, and then starting an index cluster service and an inquiry cluster service on each node; establishing, updating and deleting an index on full-text data; and finally, inquiring the index. According to the high-availability distribution type full-text index method, an inquiring and indexing process can be separated and the increment or batch type establishment of existing indexes can be simultaneously supported; the batch type indexes can be used for establishing an index for large-scale data in short time; and the increment type indexes avoid reestablishing the index. Index files can be divided into three layers of structures comprising an index file, an index fragment and an index sub-fragment, thus enhancing the expandability and the availability of the index file. According to the high-availability distribution type full-text index method, a dynamic index task configuration task is provided; and parameters in the index task are arranged to dynamically meet requirements of establishing the index by a user through different types of data.

Description

A high-availability distributed full-text index method

technical field

The invention relates to the field of information indexing and searching, and more specifically, relates to a method for constructing a distributed full-text index for massive text data and providing highly available query services.

Background technique

With the development of the Internet, especially the emergence of Web2.0, the amount of text information is increasing exponentially. Users want to be able to effectively manage massive text data, and quickly search these texts to obtain corresponding information.

The emergence of search engines such as Google, Baidu, and Bing has met the needs of users for information search. The core technology used by search engines is to collect various data information in the network through web crawlers, index these data, and then provide query services to the outside world. As the data information continues to grow, the size of the created index files also increases, resulting in the stand-alone server having been unable to save all the index files. In addition, the time spent on building a single huge index file and the time spent on executing queries on the index file are also increasing, resulting in the inability to meet the query requirements of a large number of customers. A feasible solution is to divide a single index file into multiple index shards, distribute and store each index shard on different hosts, and finally provide indexing services externally. However, through the study of several schemes adopted by the industry, it is found that these schemes have the following outstanding shortcomings:

1. The index and the query service are associated, that is, the local index file created by the index service directly provides services for the query. However, the tight coupling of query and index cluster makes the two inseparable, and the possibility of failure in terms of fault tolerance and system complexity is greatly increased.

2. Common distributed index systems adopt index sharding or index replication to increase the scalability and availability of distributed indexes. However, the sharding strategies they adopt are usually divided according to the attributes of the data, thus generating a fixed number of index shards, while the size of each index shard increases dramatically. In addition, some other distributed index systems divide the index according to a fixed size, and all index fragments need to be queried during the query process, which increases the query time and network overhead.

3. Generally speaking, distributed indexing systems are designed to meet specific needs and cannot support dynamic indexing operations on different types of data. Therefore, if you need to index a new index domain, you need to stop the distributed index cluster, redeploy the index task, and restart the index cluster to re-establish the corresponding index.

4. The current distributed index system uses multi-node backup to achieve system disaster recovery. When a node in the index cluster fails, use the backups of other nodes in the system to restore it. However, if these backup nodes fail at the same time, the index on the failed node cannot be restored normally.

Through the above analysis, we can know that there are many problems in the current distributed index system.

Contents of the invention

The purpose of the present invention is to provide a highly available distributed full-text indexing method for the deficiencies of the prior art

The purpose of the present invention is achieved through the following technical solutions: a highly available distributed full-text indexing method, the method comprising the following steps:

(1) Start the basic service system, including distributed file system, distributed columnar database and metadata directory service; the distributed file system is used to store the index files created by index nodes, and provide indexes for querying nodes in the cluster file; the distributed column database is used to provide the operation log service in the present invention, and the operation log stores various operations performed by the present invention. In addition, the distributed column database is also used to store original full-text data files, and the distributed column The database should support storing data by columns; the metadata directory service realizes the storage of metadata;

(2) Start the index cluster service and query cluster service on each node, where the index cluster service includes the index master node service and the index node service, and the query cluster service includes the query master node service and the query node service; the steps are as follows: first , the index master node and the search master node are started separately, enter the safe mode, and scan and monitor the index nodes and query node directory nodes in the metadata directory respectively, so as to manage the nodes under these directories respectively; secondly, the index node and query node are started, And register its own information on the metadata node directory. At this time, the index master node and the search master node listen to the news, and incorporate the newly added node into its own management; finally, if the node is in safe mode, the node does not change , then exit the safe mode; so far the entire process of starting the distributed full-text indexing system has been completed;

(3) Create indexes, update indexes, and delete indexes on full-text data; among them, index creation includes incremental index creation and batch index creation; this step is implemented through the following sub-steps:

The incremental index creation described in (3.1) requires the user to set the index creation mode to incremental mode in the index task definition; the process of incremental index creation includes:

(3.1.1) When data is inserted into the storage system and triggers the task of creating an incremental index, the index master node instantiates the index fragmentation strategy in the index task definition through the reflection mechanism, and determines which index segment the index of the data belongs to. After that, read the mapping relationship between the index fragment and the index node from the metadata directory, and find the index node responsible for establishing the index fragment; finally, write the node ID into the message;

(3.1.2) The index master node randomly inserts the message generated in 3.1.1 into the local message queue of any index node through remote calling;

(3.1.3) The index node reads the corresponding message. If the message belongs to the node, the index node after obtaining the message will read the data from the data storage system according to the data primary key in the message and encapsulate it into an index unit. Otherwise, the If the message does not belong to the node, then forward the message to the corresponding index node through the remote call and the node information in the message;

(3.1.3) The index node processes the message. If there is no index processor corresponding to the message, create an index processor and instantiate the indexer according to the index task definition; the indexer reads the index unit, and according to the index task in the index task Predefine the index field, index the data stored in the index unit, record the index fragment ID and operation message in the operation log 1, and record the data primary key and the name of the corresponding index sub-shard in the operation log 2;

(3.1.4) If the conditions for uploading the index sub-shard file are met at this time and the merge thread in the index processor is not running at this time, then stop the indexer on the current index sub-shard file and create a new file directory as the new For the index directory of the indexer, go to step 3.1.3; upload the index file thread and go to step 3.1.5; otherwise, still build the index under the local index sub-shard directory, go to step 3.1.3;

(3.1.5) Rename the old index sub-shard file to a merged file, set the flag that the index is being merged, and start an index file merge thread on the merged file directory to merge the index files; the upload index file thread will merge The result of uploading to the index sub-shard file directory in the distributed file system; and inserting the operation message of adding a new index sub-shard to the Master_queue in the metadata directory;

(3.1.6) If the upload is successful, delete the index operation log 1 corresponding to the index fragment, otherwise skip to step 3.1.5 and repeat the process of uploading the index fragment;

(3.2) The process of updating the index includes:

(3.2.1) The operation of updating data triggers the task of updating the index. The index master node determines which index shard the data belongs to according to the index sharding strategy, and reads the mapping between the index shard and the index node from the metadata directory Relationship, find the corresponding index node;

(3.2.2) The index master node passes the generated update index message to the local message queue of any index node through remote calls, and the index node reads the corresponding update message from it. If the message does not belong to the current node, then the node according to The node information in the message forwards the message to the real processing node;

(3.2.3) After obtaining the update message, the index node will search for the corresponding index sub-shard in the operation log 2 of the index shard according to the information in the message and the data ID, and write the index sub-shard ID into the message , if the index sub-shard is on the local disk at this time, go to step 3.2.4; otherwise, go to step 3.2.5;

(3.2.4) According to the document primary key of the data, if the index of the data is in the local index sub-shard at this time, directly update the corresponding index; and record the index shard ID and corresponding message in operation log 1 , turn to step 3.2.2;

(3.2.5) Determine whether the number of messages in the current update cache queue reaches the threshold, if not, store the message in the queue, and return to step 3.2.2; otherwise, go to step 3.2.6;

(3.2.6) Hand over all the messages in the update cache queue to the processing queue, start the update thread, and the update thread processes each message, and generates index sub-shards and indexes according to the correspondence between each data primary key and index sub-shard The one-to-many mapping relationship of the primary key. After that, the index sub-shard is downloaded from the distributed file system to the local disk, and the index processor and the corresponding indexer are instantiated on the shard. According to the data primary key in the message, the distributed Read data in the columnar database, encapsulate it into an index unit, and perform an update index operation. After success, record the update operation performed by the index sub-shard in operation log 1;

(3.2.7) When all update operations in the index sub-shard are completed, close the indexer on the directory, and upload the index file in the update directory to the directory of the distributed index sub-shard; delete the operation log 1 For the operation records related to the index sub-shard, the same operation is repeated until the mapping relationship is traversed;

(3.3) The process of deleting an index includes:

(3.3.1) The operation of deleting data triggers the task of deleting the index. The index master node determines which index shard the data belongs to according to the index sharding strategy, and reads the mapping between the index shard and the index node from the metadata directory Relationship, find the corresponding index node;

(3.3.2) The index master node passes the generated delete index message to the local message queue of any index node through remote calls, and the index node reads the corresponding delete message from it. If the message does not belong to the current node, then the node according to The node information in the message forwards the message to the real processing node;

(3.3.3) The index node after obtaining the deletion message will search for the corresponding index sub-shard in the operation log 2 of the index shard according to the information in the message and the data ID, and write the index sub-shard ID into the message , if the index sub-shard is on the local disk at this time, go to step 3.3.4; otherwise, go to step 3.3.5;

(3.3.4) According to the document primary key of the data, if the index of the data is in the local index sub-shard at this time, directly delete the corresponding index; and record the index shard ID and corresponding message in operation log 1 , turn to step 3.3.2;

(3.3.5) Determine whether the number of messages in the current deletion cache queue reaches the threshold, if not, store the message in the queue, and return to step 3.3.2; otherwise, go to step 3.3.6;

(3.3.6) Hand over all the messages in the deletion cache queue to the processing queue, start the deletion thread, and the deletion thread processes each message, and generates index sub-shards and indexes according to the correspondence between each data primary key and index sub-shard The one-to-many mapping relationship of the primary key. After that, the index sub-shard is downloaded from the distributed file system to the local disk, and the index processor and the corresponding indexer are instantiated on the shard. According to the data primary key in the message, the distributed Read the data in the columnar database, encapsulate it into an index unit and execute the delete index operation, and record the delete operation performed by the index sub-shard in the operation log 1 after success;

(3.3.7) When all deletion operations in the index sub-shard are completed, close the indexer on the directory, and upload the index file in the deleted directory to the directory of the distributed index sub-shard; delete operation log 1 For the operation records related to the index sub-shard, the same operation is repeated until the mapping relationship is traversed;

The batch indexing mode described in (3.4) is when the user defines a batch indexing mode in the index task definition. At this time, the index master node converts the mode into a batch indexing mode. The batch indexing method specifically includes the following steps:

(3.4.1) Call the MapReduce framework, call the index fragmentation strategy in the Map stage, and divide the data into n data sets; then divide the data in each data set randomly into several index nodes;

(3.4.2) Each index node creates index sub-shards according to the index shards it belongs to. Each index sub-shard is divided according to size. When the size of the index sub-shard exceeds the threshold, a new index sub-shard is created. index sub-shards;

(3.4.3) In the Reduce phase, each index node merges its own index sub-shard files into the corresponding index shard files, and will generate n index shard files;

(3.4.4) After completing the batch index building task, the index master node writes the operation of loading index fragments under the Master_queue under the metadata directory; the query master node allocates the index fragments through the node selection strategy according to this operation Give several query nodes, and write the operation of loading index fragments in the Node_queue under the metadata directory, each corresponding query node executes this operation, and loads the corresponding index;

(4) Query the index, including the following sub-steps:

(4.1) The query client generates a distributed query according to the query content, analyzes the query, and generates the corresponding analysis result;

(4.2) Afterwards, the query client uses the results generated in step 4.1 to call the local query node selection strategy; this strategy will remotely call the index fragment selection strategy on the query master node, according to the query node and index partition stored in the metadata directory. segment information to determine which index shards the query belongs to. Since an index shard has multiple index sub-shards, according to the mapping relationship between index sub-shards and query nodes in the Select the query node with the lightest load in , and generate a query node list corresponding to the query; finally, return the list to the query client;

(4.3) The query client obtains the query node list, and issues query requests to these query nodes in parallel through the method of remote calling;

(4.4) The query node executes the query request, queries the corresponding results on different index fragments through a thread pool, and returns the results to the query client;

(4.5) After the client obtains the query results of each query node, it merges the query results to complete the entire query process.

The beneficial effect of the present invention is that, firstly, the present invention separates the query and index process, making the two mutually transparent. The advantage is that the index cluster can support incremental or batch indexing at the same time. Batch indexing can meet the demand for indexing large-scale data in a short time. Incremental indexing can avoid rebuilding the established index files. On the other hand, querying the cluster only needs to read index files from the distributed file system, and does not require how the index files are created. Secondly, the present invention divides the index file into a three-layer structure, and further divides it into index sub-shards on the basis of index fragments, so as to enhance the expansibility of the index files. In the query cluster, the index sub-shard is used as a unit, and the copy of the index sub-shard is copied to multiple query nodes to improve the availability of index files. Thirdly, the present invention provides a dynamic indexing task configuration service, by setting parameters in the indexing task, it can meet user's requirements for indexing different data. Finally, the present invention records a log of each operation, and when a node in the system fails, all backup operations in the node can be restored according to the log.

Description of drawings

Figure 1 is an architecture diagram of a distributed full-text indexing system;

Fig. 2 is an organizational structure diagram of an index file of a distributed full-text indexing system;

Fig. 3 is a distribution diagram of index files in the distributed full-text search system;

Fig. 4 is a metadata directory structure diagram of the distributed full-text indexing system;

Fig. 5 is a flow chart of starting the index master node of the distributed full-text index system;

Fig. 6 is a flow chart of starting an index node of the distributed full-text indexing system;

Fig. 7 is a flow chart of starting the query master node of the distributed full-text indexing system;

Fig. 8 is a flow chart of starting the query node of the distributed full-text index system;

Fig. 9 is a flow chart of building an index in the distributed full-text indexing system;

Fig. 10 is a flowchart of deleting and updating indexes in the distributed full-text indexing system;

Fig. 11 is a flow chart of executing a query by the distributed full-text indexing system;

Fig. 12 is a schematic diagram of batch indexing in the distributed full-text indexing system.

Detailed ways

The main purpose of the present invention is to propose a method for establishing a distributed full-text index system. The distributed full-text indexing system provides massive text indexing and query services to the outside world. The present invention will be fully and detailedly described below with reference to the accompanying drawings. As shown in FIG. 1 , a distributed full-text index system can be constructed through a high-availability distributed full-text index method used in the present invention. A complete distributed full-text indexing method should consist of the following steps:

1. Start the basic service system, including distributed file system, distributed columnar database and metadata directory service. The distributed file system can use Hadoop's distributed file system HDFS, the distributed database can use HBase, and the metadata directory service can be implemented using Zookeeper. All the above-mentioned systems can be replaced by other systems with the same functions.

1.1. The distributed file system is used to store the index files created by the index nodes, and at the same time provide the index files for the query nodes in the query cluster. The distributed file system should support the storage of massive data, ensure data security and availability, and prevent data loss due to system failure.

1.2. The distributed columnar database is used to provide the operation log service in the present invention, and the operation log stores various operations performed in the method adopted in the present invention. In addition, distributed columnar databases are used to store raw full-text data files. A distributed columnar database should support storing data in columns. The operation log includes operation log 1 and operation log 2. The formats of the two are shown in the table below:

Table 1: Manipulation log 1

Index shard ID message 1 message 2 … message i … news n

Table 2: Manipulation log 2

data primary key Index sub-shard ID

1.2.1. The operation log 1 described above is used for system disaster recovery. The "index fragment ID" indicates the name of the index fragment. The present invention divides the index file into a three-layer structure, as shown in FIG. 2 . In the figure, a specific index file is composed of a specific number of index shard files. Specifically, which index shard each data belongs to can be determined according to a specific index sharding strategy. The present invention provides some index sharding strategies by default, such as dividing according to data attributes, including taking modulo according to time and document ID values, etc. Users can also customize specific index sharding strategies to meet the needs of data partitioning in different application scenarios. Each index shard file can be divided into index sub-shards according to the threshold set by the user. The distribution of three different index files in the system is shown in Figure 3.

The "message i" mentioned above contains the primary key ID and operation type of the data stored in the distributed columnar database. The operation types include: create, update and delete index operations. The index shard ID or index sub-shard ID records which index shard or sub-shard the data belongs to. The message format is shown in the table below:

data primary key operation type Index shard ID/index sub-shard ID

1.2.2. The operation log 2 is used to support index update and delete operations, see 3.2 for the specific process.

1.3. The present invention realizes the storage of metadata through the metadata directory service. The invention stores all the metadata required in the system on the metadata directory, and the metadata directory ensures the security and high availability of the metadata. The metadata directory structure is shown in Figure 4.

1.3.1. The Index directory shown serves the index cluster, and contains: Master, Node2shard, Shard2node, Live_nodes and other directories.

a. The IP address of the current index master node and the IP address of the backup index master node are saved in the shown Master directory. Once a new index master node is started, it is judged whether there is already an index master node in the directory, and if not, the IP address of the node is recorded under the directory and marked as the master node. If it exists, it will monitor the changes of the index master node. If the index master node fails, it will register itself as a new index master node and provide external services.

b. The indicated Live_nodes directory saves the IP address of the currently working index node.

c. The shown Node2shard directory saves the index shard names processed in each normal index node.

d. The shown Shard2node directory stores the one-to-one mapping relationship between index shards and index nodes.

1.3.2. The Search directory shown in the figure serves the query cluster, and contains: Master, Node, Node2shard, Shard2node, Work, Definition and other directories.

a. The IP address of the current query master node and the IP address of the backup query master node are stored in the shown Master directory. The process is the same as a in 1.3.1.

b. The shown Node2shard directory saves the names of the index shards contained in each normal query node.

c. The shown Shard2node directory stores the one-to-many mapping relationship between index shards and query nodes.

d. The shown Node directory saves the metadata information of the query node, and the status information of each node, such as CPU usage, memory space and so on.

e. The shown Work directory saves the query master node and query node operation messages. The query master node and the query node listen to the Master_queue and Node_queue directories under the current directory respectively. When the index node successfully uploads, updates or deletes an index sub-shard, it will write a new node on the Master_queue node under the directory, which contains a corresponding operation message, including adding a new index sub-shard, Overload index subshards, delete index subshards, etc. The query master node generates query node operation messages based on these operation messages, and stores these messages on the Node_queue node. Query nodes listening to this node will be notified and act accordingly.

1.3.4. The Index_metadata directory stores the metadata information of the index, including: the name of the index, the path where the index file is stored, the number of index file copies, the index fragments belonging to the index, whether there is an error in deploying the index file, etc. information.

1.3.5. The Version directory mainly stores some version information of the distributed full-text index system for verification and to prevent compatibility issues between versions.

1.3.6. The Definition directory, which saves user index task definitions. At any time when the system is running normally, users can index different data by adding index task definitions. Specifically, each indexing task has a node with the same name under the directory, and an indexing task definition file in XML format is stored in the node and serialized into binary data. This file contains information such as the name of the index task, the name of the database table that stores the data, the name of the index fragmentation strategy, the size of the index sub-shard, and the update time of the index sub-shard. At the same time, it also includes the index building mode, which currently supports batch and incremental index creation. The system performs indexing operations according to the indexing task definition.

2. The present invention needs to start index cluster service and query cluster service on each node, wherein index cluster service includes index master node service and index node service, and query cluster service includes query master node service and query node service.

2.1. The start-up process of the distributed full-text indexing system includes the following steps: first, the index master node and the search master node start respectively, enter the security mode, and scan and monitor the index node and the query node directory node in the metadata directory respectively, To manage the nodes under these directories separately. Secondly, the index node and the query node start, and register their own information on the metadata directory. At this time, the index master node and the search master node listen to the message, and incorporate the newly added node into their own management. Finally, if the node is no longer changing while in safe mode, exit safe mode. So far, the process of starting the entire distributed full-text indexing system has been completed.

2.2. The start of the index master node needs to go through the following process (as shown in Figure 5): first, judge whether it is in the initialization phase, if not, start the initialization process. Second, the path created on the metadata directory, including the Node2shard, Shard2node, and Live_node directories. Finally, register a listener on the Definition directory in the metadata directory to monitor whether a new indexing task is generated to complete the startup process.

2.3. The startup of the index node needs to go through the following processes (as shown in Figure 6): First, initialize the information of the index node, the local message queue, and start each index processor, and it is necessary to start the thread for deleting and updating the index and Relevant message queue, update the information of the local index shard, find the latest index sub-shard from HDFS, and create a new index file directory on the shard directory. Second, check whether there are some old index files locally, and upload them to HDFS if they exist. Finally, for different indexing tasks, initialize the corresponding indexer for indexing.

The index processor is a unit for establishing, updating and deleting an index, which is realized by calling an indexer. The indexer can be implemented by using a commonly used full-text indexing method, such as Lucene. This method should support index creation, update, deletion, and merge operations.

2.4. The start-up process of the query master node includes (as shown in Figure 7): firstly, start the query node selection strategy, the purpose is to select several node lists according to the query, and these nodes perform real query operations. This strategy can be realized by using polling algorithm combined with node load information. Secondly, start the query node information manager, which is used to count the load information of the query nodes and serve the query node selection strategy. Finally, start the query node manager for managing query nodes.

The process of starting the query manager includes: first judging whether the manager is in a closed state, and throwing an exception if it is in a closed state. Secondly, it is selected whether the node can become the master query node manager. If it becomes the master node, the node is registered, and each query node is notified to start the query node management process. Finally, according to the index nodes that are currently active, publish the message of monitoring whether the local index files on each query node are complete or not, and release the information of deleting expired index files. Each query node completes the corresponding operation according to the information.

2.5. The start-up process of the query node includes (as shown in FIG. 8 ): firstly, start the local index slice management program to manage the existing local index slices. Second, start the query node monitor to monitor the status of the query node, including: CPU time, memory swap rate, size of empty memory, total size of physical memory, and other information, and record these information in the metadata directory. Again, redeploy each existing index shard, and start the corresponding query service on each index shard. Finally, start the query node message queue, waiting for the message of querying the master node.

3. The highly available distributed full-text indexing method proposed by the present invention supports creating indexes on full-text data. The invention supports incremental index creation and batch index creation, and also supports update and delete operations on index files. Before creating an index, the user needs to define an index task definition and upload the file to the metadata directory. When the index master node detects that the directory node changes, it will create an index task. This task is used to process the messages corresponding to the data in the index task definition.

3.1. The incremental index creation described above requires the user to set the index creation mode to incremental mode in the index task definition. The process of incrementally creating an index includes (as shown in Figure 9):

3.1.1. When data is inserted into the storage system, the task of incremental indexing is triggered. The index master node instantiates the index sharding strategy in the index task definition through the reflection mechanism, and determines which index shard the index of the data belongs to. Afterwards, the mapping relationship between the index fragment and the index node is read from the metadata directory, and the index node responsible for establishing the index fragment is found. Finally, write the node ID into the message.

3.1.2. The index master node randomly inserts the message generated in 3.1.1 into the local message queue of any index node through remote calling.

3.1.3 The index node reads the corresponding message. If the message belongs to the node, the index node after obtaining the message will read the data from the data storage system according to the data primary key in the message and encapsulate it into an index unit. Otherwise, the message does not belong to the node, then the message is forwarded to the corresponding index node through the remote call and the node information in the message.

3.1.3. The index node processes the message. If there is no index processor corresponding to the message, create an index processor and instantiate the indexer according to the index task definition. The indexer reads the index unit, and indexes the data stored in the index unit according to the predefined index field in the index task. At the same time, record the index fragment ID and operation message in operation log 1, and record the data primary key and the name of the corresponding index sub-shard in operation log 2.

3.1.4. If the conditions for uploading the index sub-shard file are met at this time and the merge thread in the index processor is not running at this time, then stop the indexer on the current index sub-shard file and create a new file directory as a new The index directory of the indexer, go to 3.1.3. Upload the index file thread and execute 3.1.5. Otherwise, still build the index under the local index sub-shard directory, go to 3.1.3.

3.1.5. Rename the old index sub-shard file to a merged file, set the flag that the index is being merged, and start an index file merge thread on the merged file directory to merge the index files. The upload index file thread uploads the merged result to the index sub-shard file directory in the distributed file system. And insert the operation message of the new index sub-shard into the Master_queue in the metadata directory.

3.1.6. If the upload is successful, delete the index operation log 1 corresponding to the index fragment, otherwise skip to 3.1.5 and repeat the process of uploading the index fragment.

3.2. The process of updating the index includes (as shown in Figure 10):

3.2.1. The operation of updating data triggers the task of updating the index. The index master node determines which index shard the data belongs to according to the index sharding policy, reads the mapping relationship between the index shard and the index node from the metadata directory, and finds the corresponding index node.

3.2.2. The index master node transmits the generated update index message to the local message queue of any index node through remote calls. The index node reads the corresponding update message from it. If the message does not belong to the current node, then the node forwards the message to the real processing node according to the node information in the message.

3.2.3. After obtaining the update message, the index node will search for the corresponding index sub-shard in the operation log 2 of the index slice according to the information in the message and the data ID. Write the index subshard ID into the message. If the index sub-shard is on the local disk at this time, go to step 3.2.4. Otherwise, go to 3.2.5.

3.2.4. According to the document primary key of the data, if the index of the data is in the local index sub-shard at this time, directly update the corresponding index. And record the index fragment ID and the corresponding message in the operation log 1, go to step 3.2.2.

3.2.5. Determine whether the number of messages in the current update cache queue reaches the threshold, if not, store the message in the queue, and return to 3.2.2. Otherwise, go to step 3.2.6.

3.2.6. Hand over all the messages in the update cache queue to the processing queue, and start the update thread. The update thread processes each message, and generates a one-to-many mapping relationship between index sub-shards and index primary keys according to the correspondence between each data primary key and index sub-shard. Thereafter, the index sub-shards are downloaded from the distributed file system to the local disk. Instantiate the index processor and the corresponding indexer on the shard. Read data from the distributed columnar database according to the data primary key in the message, encapsulate it into an index unit and perform an update index operation. After success, record the update operation performed by the index sub-shard in operation log 1.

3.2.7. When all update operations in the index sub-shard are completed, close the indexer on the directory, and upload the index file in the update directory to the directory of the distributed index sub-shard. Delete the operation records related to the index sub-shard in operation log 1. The same operation is repeated until the mapping relationship is traversed.

3.3. The process of deleting an index is the same as that of updating an index, and will not be described again.

3.4. The batch indexing mode mentioned is when the user defines a batch indexing mode in the indexing task definition. At this point, the index master node converts the mode to a batch index mode. The batch indexing method specifically includes the following steps (as shown in Figure 12):

3.4.1. Call the MapReduce framework, call the index fragmentation strategy in the Map stage, and divide the data into n data sets. Then randomly divide the data in each data set into several index nodes.

3.4.2. Each index node creates index sub-shards for the divided data according to the index shards it belongs to. Each index sub-shard is divided by size. When the size of the index sub-shard exceeds the threshold, a new index sub-shard is created.

3.4.3. In the Reduce phase, each index node merges its own index sub-shard files into the corresponding index shard files, and will generate n index shard files.

3.4.4. After the batch index building task is completed, the index master node writes the operation of loading index fragments under the Master_queue in the metadata directory. According to this operation, the query master node assigns index fragments to several query nodes through the node selection strategy, and writes the operation of loading index fragments in the Node_queue under the metadata directory. Each corresponding query node performs this operation and loads into the corresponding index.

4. The method adopted in the present invention also supports query operations on indexes, and the query process is shown in FIG. 11 . The query steps include:

4.1. The query client generates a distributed query based on the query content, analyzes the query, and generates the corresponding analysis result.

4.2. Afterwards, the query client uses the results generated in 4.1 to call the local query node selection strategy. This strategy will remotely invoke the index shard selection strategy on the query master node, and determine which index shards the query belongs to based on the information of the query node and index shards saved in the metadata directory. Since an index shard has multiple index sub-shards, according to the mapping relationship between index sub-shards and query nodes in the metadata directory, load balancing technology is used to select the query node with the lightest load from these nodes, and generate the same query The list of corresponding query nodes. Finally, the list is returned to the querying client.

4.3. The query client obtains the query node list, and issues query requests to these query nodes in parallel through the method of remote calling.

4.4. The query node executes the query request, queries the corresponding results on different index fragments through a thread pool, and returns the results to the query client.

4.5. After the client obtains the query results of each query node, it merges the query results to complete the entire query process.

5. Since each node in the system is invisible to each other, it is necessary to provide a highly reliable message communication system to ensure the execution of each process. The message communication includes index cluster message communication, index cluster and query cluster message communication, and query cluster message communication.

5.1. The above-mentioned index cluster message communication mainly occurs between the index master node and the index node. Messages mainly include the following centralized types: new, update, delete index and mode switching messages.

5.1.1. When in the incremental index mode, whenever data enters the system, the index master node determines which index shard the data belongs to according to the index sharding strategy combined with the index shard distribution information, and records the index shard in the message. slice information, and finally put the message in the message queue of the index node.

5.1.2. When the index cluster changes from incremental to batch state, the index master node will add a mode switching message to the index node message queue. When the index node gets this message, it will notify the relevant nodes to suspend the current indexing task. The system then enters bulk indexing mode.

5.2. The above-mentioned index and query cluster message communication occurs when: when the index cluster adds, updates, deletes and merges the index files in the distributed file system, it needs to notify each relevant query node in the query cluster to update local index file. When these operations occur, the index node inserts an operation message into the Master_queue of the query master node in the metadata directory, and the query master node will receive the notification and process the message. The types of messages between the two include: check, deploy, add, reinitialize, reload, delete index and index fragmentation, etc.

5.3. The query cluster message communication is mainly caused by the message communication between the index and the query cluster. When querying the master node to obtain a new message, the query master node will parse the message, generate multiple query node messages, and insert the message into the Node_queue under the metadata directory. When the query node receives the message, it will execute the corresponding task. The message types between the query node and the master node include: deploying index shards, loading index shards, updating index shards, deleting index shards and other information.

Claims (1)

1. a high-available distributed full-text index method, is characterized in that, the method comprises the following steps:

(1) start basic service system, comprise distributed file system, distributed column formula database and metadata catalog service; The index file that described distributed file system is set up for storing index node, simultaneously for the node in inquiry cluster provides index file; Described distributed column formula database is used for providing Operation Log service, the operations that storage is carried out, in addition, distributed column formula database is also for storing original full-text data file, distributed column formula database should support that metadata catalog service realizes the storage of metadata by row storage;

(2) on each node, start index cluster service and inquiry cluster service, wherein, index cluster services package is drawn together the service of index host node and index node service, and inquiry cluster services package is drawn together inquiry host node service and query node service; This step is specially: first, index host node and search host node start respectively, enter safe mode, and scan respectively and monitor index node and query node directory node in metadata catalog, to manage respectively the node under these catalogues; Secondly, index node and query node start, and the information registering of self is arrived on metadata node catalogue, and now index host node and search host node listen to message, include the node newly adding in self-management; Finally, if under safe mode, node no longer changes, and exits so safe mode; So far completed the process that whole distributed full-text index system starts;

(3) on full-text data, create index, upgrade index, delete index; Wherein, create index and comprise that increment type creates index and batch type creates index; This step realizes by following sub-step:

(3.1) described increment type creates index, and the pattern that needs user to create index in index task definition is set to increment type pattern; The process that increment type creates index comprises:

(3.1.1) when data are inserted in storage system, trigger increment and set up the task of index, index host node is by the index stripping strategy in reflex mechanism instantiation index task definition, which index burst the index of judging these data belongs to, after this mapping relations that read index burst and index node from metadata catalog, find the index node of being responsible for setting up this index burst; Finally node ID is written in message;

(3.1.2) message that index host node generates 3.1.1 is filled in this message in the local message queue of any one index node at random by the mode of far call;

(3.1.3) index node reads corresponding message, if this message belongs to this node, the index node obtaining so after message can be according to the data major key sense data from data-storage system in message, be packaged into indexing units, otherwise this message does not belong to this node, by the nodal information in far call and message, this message is forwarded to manipulative indexing node so;

(3.1.4) index node is processed this message, if there is no, with this message manipulative indexing processor, creates index processor, and according to index task definition instantiation index; Index reads in indexing units, according to the predefine index territory in index task, the data that are stored in indexing units are carried out to index, simultaneously recording indexes burst ID and operation information in Operation Log 1, and in Operation Log 2 name of record data major key and the corresponding sub-burst of index;

If (3.1.5) now meet the condition that the sub-slicing files of index uploads and now merge thread not in operation in this index processor, stop so the index on the sub-slicing files of current index, a newly-built file directory is as the index list of new index, go to step 3.1.3, and upload index file thread and perform step 3.1.6; Otherwise still index is built under the sub-burst catalogue of local index, gone to step 3.1.3;

(3.1.6) sub-old index slicing files is renamed as merged file, the mark that index is merging is set, and in merged file catalogue, start an index file merging thread, index file is merged; Upload index file thread the result of merging is uploaded to the sub-slicing files catalogue of index in distributed file system; And the operation information of the newly-increased sub-burst of index is inserted on the Master_queue in metadata catalog;

If (3.1.7) uploaded successfully, so delete the index operation daily record 1 corresponding with this index burst, otherwise jump to step 3.1.6 and repeat to upload the process of index burst;

(3.2) process of renewal index comprises:

(3.2.1) more the operation of new data triggers the task of upgrading index, index host node judges according to index stripping strategy this attribution data is in which index burst, and read the mapping relations of index burst and index node from metadata catalog, find corresponding index node;

(3.2.2) index host node passes to the renewal index messages of generation by far call the local message queue of any index node, this index node therefrom reads corresponding updating message, if this message does not belong to present node, this node is forwarded to real processing node according to the nodal information in message by message so;

(3.2.3) index node obtaining after updating message can be according to the information in message, in the Operation Log 2 of this index burst, search the sub-burst of corresponding index according to data ID, sub-index burst ID is write in message, if now the sub-burst of index, on local disk, goes to step 3.2.4; Otherwise go to step 3.2.5;

(3.2.4) according to the document major key of these data, if now the index of these data, among the sub-burst of local index, directly upgrades corresponding index; And in Operation Log 1 recording indexes burst ID and corresponding message, go to step 3.2.2;

(3.2.5) judge whether the message number in current renewal buffer queue arrives threshold value, if do not arrived, deposit this message in this queue, return to step 3.2.2; Otherwise go to step 3.2.6;

(3.2.6) all give processing queue by the message of upgrading in buffer queue, start more new thread, upgrade each message of thread process, according to the corresponding relation of each data major key and the sub-burst of index, the one-to-many mapping relations of the sub-burst of generating indexes and index major key, after this, from distributed file system, sub-index burst is downloaded to local disk, on this burst, example dissolves index processor and corresponding index, according to the reading out data from distributed column formula database of the data major key in message, be packaged into indexing units and carry out and upgrade index operation, after success, in Operation Log 1, record the renewal operation that the sub-burst of this index is carried out,

(3.2.7) when completing all renewals operation in the sub-burst of this index, be closed in the index in this catalogue, and the index file in new directory be more uploaded to the catalogue of the sub-burst of distributed index; The operation note relevant to the sub-burst of this index in deletion action daily record 1, same operation repeats until traveled through mapping relations;

(3.3) process of deletion index comprises:

(3.3.1) operation of deleting data triggers the task of deleting index, index host node judges according to index stripping strategy this attribution data is in which index burst, and read the mapping relations of index burst and index node from metadata catalog, find corresponding index node;

(3.3.2) index host node passes to the deletion index messages of generation by far call the local message queue of any index node, this index node therefrom reads corresponding deletion message, if this message does not belong to present node, this node is forwarded to real processing node according to the nodal information in message by message so;

(3.3.3) obtaining the index node of deleting after message can be according to the information in message, in the Operation Log 2 of this index burst, search the sub-burst of corresponding index according to data ID, sub-index burst ID is write in message, if now the sub-burst of index, on local disk, goes to step 3.3.4; Otherwise go to step 3.3.5;

(3.3.4) according to the document major key of these data, if now the index of these data, among the sub-burst of local index, is directly deleted corresponding index; And in Operation Log 1 recording indexes burst ID and corresponding message, go to step 3.3.2;

(3.3.5) judge whether the message number in current deletion buffer queue arrives threshold value, if do not arrived, deposit this message in this queue, return to step 3.3.2; Otherwise go to step 3.3.6;

(3.3.6) all give processing queue by the message of deleting in buffer queue, start and delete thread, delete each message of thread process, according to the corresponding relation of each data major key and the sub-burst of index, the one-to-many mapping relations of the sub-burst of generating indexes and index major key, after this, from distributed file system, sub-index burst is downloaded to local disk, on this burst, example dissolves index processor and corresponding index, according to the reading out data from distributed column formula database of the data major key in message, be packaged into indexing units and carry out and delete index operation, after success, in Operation Log 1, record the deletion action that the sub-burst of this index is carried out,

(3.3.7) when completing all deletion actions in the sub-burst of this index, be closed in the index in this catalogue, and the index file in deltreeing be uploaded to the catalogue of the sub-burst of distributed index; The operation note relevant to the sub-burst of this index in deletion action daily record 1, same operation repeats until traveled through mapping relations;

(3.4) described batch type indexing model is to set up indexing model when user has defined batch type in index task definition, and now, pattern is converted to batch type indexing model by index host node, and batch type indexing means specifically comprises following steps:

(3.4.1) call MapReduce framework, call index stripping strategy in the Map stage, data are divided into n data acquisition; Then by the data random division in each data acquisition to several index nodes;

(3.4.2) each index node, to the data of dividing according to the index burst under it, is set up respectively the sub-burst of index, and the sub-burst of each index is divided according to size, when the sub-burst size of index exceedes threshold value, creates the sub-burst of new index;

(3.4.3), in the Reduce stage, the corresponding index slicing files that each index node merges to sub-index in self slicing files, will generate n index slicing files;

(3.4.4) when completing after batch type indexes task, index host node writes the operation that is written into index burst under the Master_queue under metadata catalog; Inquiry host node is distributed to several query node by node selection strategy by index burst according to this operation, and in the Node_queue under metadata catalog, write the operation that is written into index burst, each corresponding query node is carried out this operation, is written into corresponding index;

(4) search index, comprises following sub-step:

(4.1) inquiring client terminal generates distributed query according to query contents, and this inquiry is resolved, and produces corresponding analysis result;

(4.2) after this inquiring client terminal utilizes the result producing in step 4.1, calls local query node selection strategy; This strategy can be inquired about the index burst selection strategy on host node by far call, the query node of preserving according to metadata catalog and the information of index burst, determine which index burst this inquiry belongs to, because an index burst has the sub-burst of multiple indexes, therefore according to the mapping relations of the sub-burst of index in metadata catalog and query node, utilize load-balancing technique from these nodes, to select the lightest query node of load, generate the query node list corresponding with this inquiry; Finally, this list is returned to inquiring client terminal;

(4.3) inquiring client terminal obtains this query node list, by the method for far call, walks abreast to these query node releasing inquiry requests;

(4.4) query node is carried out this inquiry request, inquires about corresponding result by a thread pool on different index bursts, and result is returned to inquiring client terminal;

(4.5) after client is obtained the Query Result of each query node, Query Result is carried out to union operation, complete whole query script.