DYNAMIC RESOURCE ALLOCATION FOR REAL-TIME SEARCH
Systems and methods are disclosed for utilizing an ingested data buffer operating according to a publish-subscribe messaging model as an intake mechanism for a query system. Data from various sources can be placed into the data buffer according to different topics. Indexers can subscribe to these topics in order to ingest the data into the system for long-term storage and later search. In addition, worker nodes may directly subscribe to the topics to enable continuous or streaming searching of the data, without delays that may be caused by ingestion of the data at an indexer. When a request for a streaming search is received, a query coordinator can determine a number of message queues on the data buffer that contain potentially relevant messages. The query coordinator can then dynamically allocate partitions operating on worker nodes to retrieve and intake messages from the message queues into a phased search process.
1 . A method comprising:
receiving a query including criteria for identifying search results from a set of data comprising messages published to a publish-subscribe messaging system;
defining, by the data index and query system, a query processing scheme for obtaining and processing the set of data, the query processing scheme including a dynamic allocation of partitions to process the messages to identify the search results, wherein defining the query processing scheme comprises:
determining a number of messages queues on the publish-subscribe messaging system in which the messages are placed;
dynamically allocating individual queues of the message queues to individual partitions, the individual partitions to repeatedly collect messages from their allocated individual over a plurality of time windows, and at the end of each time window, cause the messages received during that time window to be processed by a processing partition; and
executing the query based on the query processing scheme.
2 . The method of claim 1 , wherein each time window is a predetermined length of time.
3 . The method of claim 1 , wherein the partitions are implemented by one or more worker node computing devices, and wherein individual partitions correspond to at least one of a processor of the one or more worker node computing devices or an execution environment on the one or more worker node computing devices.
4 . The method of claim 1 , wherein the partitions are implemented by one or more worker node computing devices, and wherein dynamically allocating individual queues of the message queues to individual partitions comprises evenly allocating the individual queues among the one or more worker node computing devices.
5 . The method of claim 1 , wherein a number of the partitions is equal to a number of the message queues, and wherein dynamically allocating individual queues to individual partitions comprises allocating a single queue to each partition.
6 . The method of claim 1 , wherein a number of the partitions is less than a number of the message queues, and wherein dynamically allocating individual queues to individual partitions comprises allocating multiple queues to each of the partitions.
7 . The method of claim 1 , wherein at least two of the partitions are implemented within a common worker node computing device.
8 . The method of claim 1 , wherein each of the partitions is implemented within a different worker node computing device.
9 . The method of claim 1 , wherein the set of data corresponds to a topic of the publish-subscribe messaging system.
10 . The method of claim 1 , wherein the set of data corresponds to a topic of the publish-subscribe messaging system, and wherein determining the number of messages queues on the publish-subscribe messaging system in which the messages are placed comprises querying the publish-subscribe messaging system for identifying information of one or more message queues corresponding to the topic.
11 . The method of claim 1 , wherein the set of data corresponds to a topic of the publish-subscribe messaging system, wherein determining the number of messages queues on the publish-subscribe messaging system in which the messages are placed comprises querying the publish-subscribe messaging system for identifying information of one or more message queues corresponding to the topic, and wherein querying the publish-subscribe messaging system for identifying information of the one or more message queues corresponding to the topic comprises instructing a partition of the partitions to query the publish-subscribe messaging system for the identifying information.
12 . The method of claim 1 , wherein defining the query processing scheme further comprises:
transmitting a request for a number of partitions used to execute the query to a workload advisor computing device;
receiving a response from the workload advisor computing define indicating a permitted number of partitions to be used to execute the query; and
identifying the partitions based at least partly on the permitted number of partitions.
13 . The method of claim 1 , wherein defining the query processing scheme further comprises:
transmitting a request for a number of partitions used to execute the query to a workload advisor computing device;
receiving a response from the workload advisor computing define indicating a permitted number of partitions to be used to execute the query; and
identifying the partitions based on at least partly the permitted number of partitions, wherein the partitions are further identified based on information indicating individual worker node computing devices that implement individual partitions of the partitions.
14 . The method of claim 1 , wherein the partitions correspond to a first set of partitions, and wherein defining the query processing scheme further comprises dynamically allocating a second set of partitions as processing partitions by parsing messages collected by the partitions of the first set of partitions to identify information responsive to the query.
15 . The method of claim 1 , wherein the partitions correspond to a first set of partitions, wherein defining the query processing scheme further comprises dynamically allocating a second set of partitions as processing partitions by parsing messages collected by the partitions of the first set of partitions to identify information responsive to the query, and wherein the second set of partitions includes at least one partition of the first set of partitions.
16 . The method of claim 1 , wherein the partitions correspond to a first set of partitions, wherein defining the query processing scheme further comprises dynamically allocating a second set of partitions as processing partitions by parsing messages collected by the partitions of the first set of partitions to identify information responsive to the query, and wherein the second set of partitions includes at least one partition of the first set of partitions and at least one other partition.
17 . The method of claim 1 , wherein the partitions correspond to a first set of partitions, wherein defining the query processing scheme further comprises dynamically allocating a second set of partitions to parse the portions of data to identify the one or more results, wherein the second set of partitions includes at least one partition of the first set of partitions and at least one other partition, and wherein the at least one other partition is identified based at least in part on identifying a worker node computing device that implements both the at least one partition of the first set of partitions and the at least one other partition.
18 . The method of claim 1 , wherein the partitions correspond to a first set of partitions, wherein defining the query processing scheme further comprises:
dynamically allocating a second set of partitions to parse the collected messages to identify one or more results; and
dynamically allocating a third set of partitions to aggregate the one or more results into an aggregate result.
19 . The method of claim 1 , wherein the partitions correspond to a first set of partitions, wherein defining the query processing scheme further comprises:
dynamically allocating a second set of partitions to parse the portions of data to identify the one or more results; and
dynamically allocating a third set of partitions to aggregate the one or more results into an aggregate result, wherein the third set of partitions is selected according to a grouping criteria of the query.
20 . The method of claim 1 further comprising identifying, from the criteria, that the query is directed to data on the publish-subscribe messaging system based at least in part on the criteria of the query.
21 . The method of claim 1 further comprising identifying, from the criteria, that the query is directed to data on the publish-subscribe messaging system based at least in part on the criteria of the query.
22 . The method of claim 1 further comprising transmitting search results generated from a collection of messages to a data destination and, subsequent to transmitting the search results search results generated from the collection of messages, transmitting an acknowledgement of each message within the collection of messages to the publish-subscribe messaging system.
23 . The method of claim 1 further comprising transmitting search results generated from a collection of messages to a data destination and, subsequent to transmitting the search results search results generated from the collection of messages, transmitting an acknowledgement of each message within the collection of messages to the publish-subscribe messaging system, wherein transmitting the acknowledgement of each message within the collection of messages to the publish-subscribe messaging system comprises instructing partitions that collected each message within the collection of messages to transmit acknowledgements for each message to the publish-subscribe messaging system.
24 . The method of claim 1 , wherein the messages are associated with a topic on the publish-subscribe messaging system, wherein the method further comprising transmitting the search to the publish-subscribe messaging system as messages for a second topic.
25 . The method of claim 1 , wherein the publish-subscribe messaging system queues the messages for delivery to indexers that index data within the messages.
26 . The method of claim 1 , wherein the publish-subscribe messaging system queues the messages for delivery to indexers that index data within the messages, and wherein the indexers place indexed data generated from the messages within a common storage also accessible to the partitions.
27 . A system comprising:
a data store including computer-executable instructions; and
one or more processors configured to execute the computer-executable instructions, wherein execution of the computer-executable instructions causes the system to:
receive a query including criteria for identifying search results from a set of data comprising a plurality of buckets of data;
define, by a data index and query system, a query processing scheme for identifying the search results from the set of data, the query processing scheme including a dynamic allocation of partitions to process the plurality of buckets to identify the search results, wherein defining the query processing scheme comprises:
determining, from metadata regarding the plurality of buckets, a subset of buckets from the plurality of buckets that potentially include the search results; and
dynamically allocating individual buckets from the subset of buckets to individual partitions, the individual partitions to obtain in parallel the individual buckets allocated to the individual partitions and to locate, within the individual buckets, portions of data that potentially include one or more results of the search results; and
execute the query based on the query processing scheme.
28 . The system of claim 27 , wherein execution of the computer-executable instructions causes the system to transmit search results generated from a collection of messages to a data destination and, subsequent to transmitting the search results search results generated from the collection of messages, transmit an acknowledgement of each message within the collection of messages to the publish-subscribe messaging system.
29 . Non-transitory computer-readable media including computer-executable instructions that, when executed by a computing system, cause the computing system to:
receive a query including criteria for identifying search results from a set of data comprising a plurality of buckets of data;
define, by a data index and query system, a query processing scheme for identifying the search results from the set of data, the query processing scheme including a dynamic allocation of partitions to process the plurality of buckets to identify the search results, wherein defining the query processing scheme comprises:
determining, from metadata regarding the plurality of buckets, a subset of buckets from the plurality of buckets that potentially include the search results; and
dynamically allocating individual buckets from the subset of buckets to individual partitions, the individual partitions to obtain in parallel the individual buckets allocated to the individual partitions and to locate, within the individual buckets, portions of data that potentially include one or more results of the search results; and
execute the query based on the query processing scheme.
30 . The non-transitory computer-readable media of claim 29 , wherein execution of the computer-executable instructions causes the computing system to transmit search results generated from a collection of messages to a data destination and, subsequent to transmitting the search results search results generated from the collection of messages, transmit an acknowledgement of each message within the collection of messages to the publish-subscribe messaging system.