WO2015116223A1

WO2015116223A1 - Execution of a query on data updates and pushing query results

Info

Publication number: WO2015116223A1
Application number: PCT/US2014/014332
Authority: WO
Inventors: Sebastien TANDEL
Original assignee: Hewlett Packard Development Co LP
Current assignee: Hewlett Packard Development Co LP
Priority date: 2014-01-31
Filing date: 2014-01-31
Publication date: 2015-08-06
Anticipated expiration: 2016-07-31

Abstract

A method includes receiving, in a system having a processing pipeline with a plurality of processing stages, data updates, wherein the processing stages perform respective different operations on the data updates received by the system, wherein the plurality of processing stages includes an ingest stage to ingest the data updates into the system, and a merging stage to merge the data updates into a data table. The method includes receiving a query with a coordinator, executing the query on the data updates prior to the data updates being merged into the data table, and asynchronously pushing query results of the query to a client device.

Description

EXECUTION OF A QUERY ON DATA UPDATES AND PUSHING QUERY

RESULTS

Background

[0001] An organization may have a relatively large amount of data that users or applications within the organization may request to perform data mining, analysis, search, or other tasks. As systems become more complex and as the amount of data increases, the ability to efficiently access data maintained by such systems has become more challenging.

Brief Description of the Drawings

[0002] Figure 1 is a block diagram illustrating a data processing system having a processing pipeline according to an example of the present techniques.

[0003] Figure 2 is a flow diagram illustrating a data processing method according to an example of the present techniques.

Detailed Description

[0004] In the following detailed description, reference is made to the

accompanying drawings which form a part hereof, and in which is shown by way of illustration specific examples in which the disclosure may be practiced. It is to be understood that other examples may be utilized and structural or logical changes may be made without departing from the scope of the present disclosure. The following detailed description, therefore, is not to be taken in a limiting sense, and the scope of the present disclosure is defined by the appended claims. It is to be understood that features of the various examples described herein may be combined, in part or whole, with each other, unless specifically noted otherwise.

[0005] A data processing system may have multiple processing stages for performing respective processing of data. After one processing stage has completed its respective processing, the processing stage may send processed data to another processing stage for further processing. In some examples, a data processing system having multiple processing stages is arranged as a processing pipeline since the multiple processing stages are arranged to sequentially apply processing of data that passes through the processing pipeline.

[0006] In the ensuing discussion, reference is made to examples applicable to a processing pipeline. However, techniques or mechanisms according to some examples may be applied to other types of data processing systems. The data processing system may be implemented with a computer system or a combination of computer systems, where each computer system may have one or multiple processors.

[0007] In some examples, a processing pipeline is configured to process data updates. Data updates may be provided from various sources. A "data update" refers to creation of data, modification of data, and/or deletion of data. Because there may be a relatively large amount of data updates to be processed by a processing pipeline, it may take a relatively long period of time before the data updates being processed by the processing pipeline are available for access by queries submitted to the processing pipeline, if queries are unable to access intermediate results of the processing pipeline.

[0008] In accordance with some examples, techniques or mechanisms are provided to obtain more timely results from the processing pipeline in response to a query. A query may be associated with a particular freshness specification, where "freshness" of data refers to how up-to-date results should be for a response to the query. In some examples or applications, a user may want a relatively quick response to a query, but the user may be willing to accept results that are out-of-date by a certain amount of time, as indicated by a freshness specification (e.g., out-of-date by 12 hours, one day, etc.). On the other hand, other users or applications (such as a virus scanning application) may want an up-to-date response regarding data in the processing pipeline, at the expense of a slower response time to a query.

[0009] Figure 1 is a block diagram illustrating a data processing system having a processing pipeline according to one example. The illustrated system includes a server system 100 having a processing pipeline 102. System 100 provides efficient query streaming for real-time analytics in a distributed database. The database executes or runs on clusters that may be composed of several tens of servers. Each of the servers is an owner of some part of the database. The database is configured or architected with a share nothing concept in mind, and the servers do not maintain any state and are coordinated by a master.

[0010] The system 100 may be configured to be responsible for collecting events occurring in a file system (e.g., new files, file deletion, file modified, new directory, etc.) and exposing it to clients for querying. The system 100 comprises a processing pipeline 102 composed of several stages, each stage representing a step, from the ingestion of the data to the merge of the data into the database, with the last step being the realization of the different indexes. Data may be made available for queries after all these steps are executed, and a polling scheme may be used for queries on the data after it has passed through the entire processing pipeline 102.

[0011] This configuration may provide for a relatively high-rate of ingestion and an efficient and fast query system with the ability to scale, but involves a trade-off. It may consume some time to take the data from the ingest stage 104 to the end of the merge stage 110, so there is an expense of waiting some time to make data available for queries (i.e., freshness). Although that time may be short enough for some use cases, it may not be for other use cases. For example, assume that a customer is repeatedly performing the same query, in a poll style, and waiting for some changes in the result set. This may have two potential negative costs: (1) In order to return the answer, the system might parse data that are not of interest for the result, which comes at a significant cost when dealing with large tables containing, for example, trillions of events; and (2) there might be no result in the answer, which might have a significant disk 10 cost. As another example, if data are merged into a new database generation in batch, this may prohibit having a fine-grained view of the dynamics of the system. Faster execution and report generation provides the ability to give insight on the dynamicity at a fine-grained level. As described in further detail below, system 100 may help address this issue and provide efficient query streaming for real-time analytics in a distributed database.

[0012] The processing pipeline 102 has an ingest stage 104, an ID (identifier) remapping stage 106, a sorting stage 108, and a merging stage 110. Although specific stages of the processing pipeline 102 are depicted in Figure 1 , it is understood that in different examples, alternative stages or additional stages may be provided in the processing pipeline 102.

[0013] Data updates from various update clients 1 2 are provided to the server system 100 for processing by the processing pipeline 102. Examples of the update clients 112 include various machines that may store data within an organization, where the machines may include electronic data processing devices such as desktop computers, notebook computers, personal digital assistants (PDAs), various types of servers (e.g., file servers, email servers, etc.), or other types of devices. The machines that comprise the update clients 112 may provide sources of data such as stock market transactions, web logs, cluster logs, e-commerce history, and so forth.

[0014] A data update that is sent or transmitted to the server system 100 may include the metadata associated with the actual data stored on the update clients 112. In such examples, the data update includes the metadata, but not the actual data. Examples of metadata include metadata computed based on content of the data, including hashes (produced by applying hash functions on actual data), term vectors (containing terms in the data), fingerprints, feature vectors, and so forth. Other examples of metadata include file system metadata, such as file owners or creators, file size and security attributes, or information associated with usage of the data, such as access frequency statistics. Alternatively, instead of just metadata, actual data may be stored in the server system 100, such as data associated with timestamps, e.g. sensor observations, log entries, transaction records, social networking messages, and so forth.

[0015] As further depicted in Figure 1 , query clients 1 18 may submit queries 120 to the server system 100. A query processing engine 130 in the server system 100 may be configured to respond to the queries 120 with responses 122 that are provided back to the query clients 118.

[0016] As further shown in Figure 1 , processors 150 are provided in the server system 100. The processors 150 may be part of one or multiple computer nodes. In some examples, the query processing engine 130 and the processing stages 104, 106, 108, and 110 may be provided on respective computer nodes.

[0017] In some examples, updates from the update client(s) 112 are applied to an "authority table" 1 4 stored in a data store or database 116 of the server system 100. An authority table 114 refers to a repository of the data that is to be stored by the server system 100, where the authority table 114 may be searched in response to a query for data. The data store 116 may store multiple authority tables 114, in some examples. More generally, the authority tables 114 are referred to as data tables, which are contained in a database.

[0018] Another type of table that may be maintained by the server system 100 is an update table, which is an intermediate table that contains additions, modifications, and/or deletions (based on the data updates received from the update clients 112) that are to be applied to an authority table 114 after processing through the processing pipeline 102. An update table has the same schema as the associated authority table, as well as additional columns to indicate the type of operation and a timestamp. The various processing stages (104, 106, 108, 110) are configured to process update tables. The update tables may be stored on nodes different from a node (or nodes) storing authority tables 114.

[0019] In some examples, multiple updates may be batched into a single self- consistent update (SCU) (more generally referred to as a "batch of updates"). Each SCU includes one or plural update tables containing update data. The SCU is applied in the server system 100 as a single atomic unit, and is not considered durable until all the individual updates in the batch (SCU) are written to stable (persistent) storage. Atomic application of data updates of an SCU to the stable storage means that all data updates of the SCU are applied or none are applied. Data updates in any one SCU are isolated from data updates in another SCU. Batching of data updates may be omitted in other examples..

[0020] The ingest stage 104 of the processing pipeline 102 batches (collects) incoming updates from update clients 112 into one or plural unsorted SCUs 105. As shown in Figure 1 , the output (105) of the ingest stage 104 is an unsorted SCU (or multiple unsorted SCUs 105). The unsorted SCU(s) 105 is (are) provided to the ID remapping stage 106, which transforms initial

(temporary) ID(s) of the SCU(s) 105 into global ID(s). Effectively, the ID remapping stage 106 maps an ID in a first space to an ID in a second space, which in some examples is a global space to provide a single, searchable ID space. The initial (temporary) IDs used by the ingest stage 104 are assigned to each unique entity (for example, file names) as those entities are processed. ID's are used in place of relatively large pieces of incoming data such as file path names, which improves query processing times and reduces usage of storage space. In addition, in some examples where the ingest stage 104 is implemented with multiple processors, temporary IDs generated by each of the processors may be remapped to the global ID space. In this way, the processors of the ingest stage 104 do not have to coordinate with each other to ensure generation of unique IDs, such that greater parallelism may be achieved.

[0021] The output of the ID remapping stage 106 includes one or plural remapped SCUs 107 (within each remapped SCU 107, an initial ID has been remapped to a global ID). Each remapped SCU 107 is provided to the sorting stage 108, which sorts one or plural update tables in the remapped SCU by one or plural keys to create a sorted SCU 09 that contains one or plural full searchable indexes (e.g. extent-based indexes). A full searchable index is an index produced from one or multiple columns (attributes) of each sorted SCU.

[0022] The sorted SCU(s) 109 is (are) provided to the merging stage 110. The merging stage 110 combines individual sorted SCUs to further improve query performance. The output of the merging stage 110 includes one or multiple merged SCUs 111. Each merged SCU 111 may also be associated with a full searchable index. The merged SCU(s) 11 is (are) merged into the authority table(s) 114. Note that there may be several types of merging— the merging stage 10 may produce merged SCUs, or alternatively, a new version of an authority table (with updates merged in).

[0023] As mentioned above, using a polling scheme for queries on data after it has passed through the entire processing pipeline 02 may provide for a high-rate of ingestion and an efficient and fast query system, but there is may be an expense of waiting some time to make data available for queries (i.e., freshness). The merge time for large systems takes time, and there may be a delay in obtaining a different answer for the same query. The stages 04, 106, and 108 may occur in milliseconds or seconds, while the merging stage 110 may take minutes or longer. In one example, system 100 may provide clients with the ability to get real-time or near real-time information from the database while minimizing (or amortizing) the resources overhead to obtain it. In one form of this example, query clients 118 register pre-configured queries with coordinator 160, as indicated at 148, and receive a stream of results through queue 144, as indicated at 146. More specifically, a query client 1 8 registers a set of queries with coordinator 160, and specifies a queue ID of a queue 144 to which the results will be returned through.

[0024] When a sort process is begun by sorting stage 108, the coordinator 160 passes the registered set of queries to the sort process for execution. While performing the sort process, the sorting stage 108 also executes the queries received from coordinator 160 on the new dataset entering into the system 100 and being sorted. In another example, merging stage 110 uses a log-structured merge (LSM) tree for merging, and the queries received from the coordinator 160 are executed between the sorting stage 108 and the merging stage 110 (i.e., before a first level (level 0) of the LSM) or at the first level (level 0) of the LSM. Results from the query execution are put into the queue 144 defined by the query client 118. In one example, the queue 144 asynchronously pushes the query results as a stream to the query client 118, as indicated at 146. By pushing the results to the query client 118, rather than relying on a polling scheme, the overall resources for answering the registered queries is decreased (or amortized over time). In one example, queue 144 uses advanced message queuing protocol (AMQP) or web sockets to asynchronously return the query results to query client 118.

[0025] A query registered with coordinator 160 may be performed in at least three different modes: (1) execute the query for all rows and return every result without sampling; (2) execute the query at a sampling frequency specified by the query client 118, and return every result; and (3) continuously execute the query, and return results at a sampling frequency specified by the query client 118. The queries registered with coordinator 160 may also be "temporal" queries that return consolidated results collected during a time window specified by the query client 118.

[0026] Since the stages 104, 106, and 108 typically process the ingested data, which is a new dataset typically containing a small quantity of data, and not in batch, the execution time for executing the queries at the sorting stage 108 is on the order of milliseconds or seconds. Thus, the query clients 118 receive the query results in a push-style in near real-time fashion. The query clients 118 are then able to consume these data at their own pace. In this manner, the system 100 enables new use cases, such as change set query types and obtaining real-time or near real-time analytics of the system, in a system that includes a file system that may scale to billions of objects. Regarding the enabling change set query types, a query may be registered that asks which files have been touched/changed since a specific date/time, and Common Internet File System (CIFS) dir/file change notifications may be plugged into or connected to the stream. Regarding the obtaining real-time analytics, the fast execution time at the sorting stage 108 enables clients to get a view of the dynamics of the system, including administrative statistical insights regarding file system activity (e.g., determining real-time changes in the system and determining the most active users in the system for the last few seconds or minutes), and the detection of a file system attack or other real-time analytics. [0027] Machine-readable instructions of modules described above (such as the query processing engine 130 and the coordinator module 160) are loaded for execution on a processor (such as a CPU). A processor may include a microprocessor, microcontroller, processor module or subsystem,

programmable integrated circuit, programmable gate array, or another control or computing device.

[0028] Data and instructions are stored in respective storage devices, which are implemented as computer-readable or machine-readable storage media. The storage media include different forms of memory including semiconductor memory devices such as dynamic or static random access memories (DRAMs or SRAMs), erasable and programmable read-only memories (EPROMs), electrically erasable and programmable read-only memories (EEPROMs) and flash memories; magnetic disks such as fixed, floppy and removable disks; other magnetic media including tape; optical media such as compact disks (CDs) or digital video disks (DVDs); or other types of storage devices. Note that the instructions discussed above may be provided on one computer-readable or machine-readable storage medium, or alternatively, may be provided on multiple computer-readable or machine-readable storage media distributed in a large system having possibly plural nodes. Such computer-readable or machine-readable storage medium or media is (are) considered to be part of an article (or article of manufacture). An article or article of manufacture may refer to any manufactured single component or multiple components. The storage medium or media may be located either in the machine running the machine- readable instructions, or located at a remote site from which machine-readable instructions may be downloaded over a network for execution.

[0029] Depending on the exact configuration and type of server system 100, the memory of system 100 may be volatile (such as RAM), non-volatile (such as ROM, flash memory, etc.), or some combination of the two. System 100 may also have additional or different features/functionality and additional or different hardware and software. For example, system 100 may also include additional storage (removable and/or non-removable) including, but not limited to, magnetic or optical disks or tape. Computer storage media includes volatile and nonvolatile, removable and non-removable media implemented in any suitable method or technology for non-transitory storage of information such as computer readable instructions, data structures, program modules or other data. The memory of system 100 is an example of computer storage media (e.g., computer-readable storage media storing computer-executable instructions that when executed by at least one processor cause the at least one processor to perform a method). Computer storage media includes RAM, ROM, EEPROM, flash memory or other memory technology, CD-ROM, digital versatile disks (DVD) or other optical storage, magnetic cassettes, magnetic tape, magnetic disk storage or other magnetic storage devices. Any such computer storage media may be part of storage system 100.

[0030] One example is directed to a data processing method. Figure 2 is a flow diagram illustrating a data processing method 200 according to one example.. Server system 100 is configured to perform method 200. At block 202 in method 200, in a system having a processing pipeline with a plurality of processing stages, data updates are received, wherein the processing stages perform respective different operations on the data updates received by the system, wherein the plurality of processing stages includes an ingest stage to ingest the data updates into the system, and a merging stage to merge the data updates into a data table. At block 204, a query is received by a coordinator, wherein the query is sent to the coordinator by a client device. At block 206, the query is executed on the data updates prior to the data updates being merged into the data table. At block 208, query results of the query are asynchronously pushed or transmitted to the client device.

[0031] In one example of method 200, the plurality of processing stages also includes a sorting stage to sort the data updates after ingestion of the data updates by the ingest stage, and wherein the merging stage merges the sorted data updates into the data table. In one form of this example, the method 200 also includes performing a sort process by the sorting stage; passing the query from the coordinator to the sort process; and executing the query with the sort process on data updates being sorted by the sort process. In another example, the method 200 includes executing the query between the sort stage and the merging stage. In another example, the merging stage uses a log-structured merge (LSM) tree for merging, and the method includes executing the query at a first level of the LSM.

[0032] Method 200 according to one example also includes storing the results of the query in a queue specified by the client device; and asynchronously pushing the results of the query as a stream from the queue to the client device. In one form of this example, the queue uses advanced message queuing protocol (AMQP) or web sockets to asynchronously push the results of the query to the client device.

[0033] In one example, method 200 includes executing the query at a sampling frequency specified by the client device; and asynchronously pushing results from each execution of the query to the client device. In another example, method 200 includes continuously executing the query; and asynchronously pushing results from the continuous execution of the query to the client device at a sampling frequency specified by the client device. The results of the query according to one example comprise consolidated results collected during a time window specified by the client device.

[0034] Another example is directed to a data processing system that includes a processing pipeline having a plurality of processing stages, wherein the processing stages perform respective different operations on data updates received by the system, wherein the plurality of processing stages includes an ingest stage to ingest the data updates into the system, and a merging stage to merge the data updates into a data table. The system includes a coordinator to receive a query from a client device and cause the query to be executed on the data updates prior to the data updates being merged into the data table. The system includes a queue to store query results of the query and asynchronously push the query results to the client device.

[0035] In one form of this example, the plurality of processing stages includes a sorting stage to sort the data updates after ingestion of the data updates by the ingest stage, wherein the merging stage merges the sorted data updates into the data table, and wherein the sorting stage executes the query on data updates being sorted by the sorting stage. The queue according to one example uses one of advanced message queuing protocol (AMQP) and web sockets to push the query results to the client device.

[0036] Yet another example is directed to a data processing system that includes a processing pipeline having a plurality of processing stages, wherein the processing stages perform respective different operations on data updates received by the system, wherein the plurality of processing stages includes an ingest stage to ingest the data updates into the system, a sorting stage to sort the data updates after ingestion of the data updates by the ingest stage, and a merging stage to merge the sorted data updates into a data table. The system includes a coordinator to receive a query from a client device and cause the query to be executed on the data updates prior to the data updates being merged into the data table, wherein the coordinator causes the query to be executed at a frequency specified by the client device. The system includes a queue to store query results of the query and asynchronously push or transmit the query results to the client device at a frequency specified by the client device.

[0037] Although specific examples have been illustrated and described herein, a variety of alternate and/or equivalent examples may be substituted for the specific examples shown and described without departing from the scope of the present disclosure. This application is intended to cover any adaptations or variations of the specific examples discussed herein. Therefore, it is intended that this disclosure be limited only by the claims and the equivalents thereof.

Claims

1. A method comprising:

receiving, in a system having a processing pipeline with a plurality of processing stages, data updates, wherein the processing stages perform respective different operations on the data updates received by the system, wherein the plurality of processing stages includes an ingest stage to ingest the data updates into the system, and a merging stage to merge the data updates into a data table;

receiving a query with a coordinator, wherein the query is sent to the coordinator by a client device;

executing the query on the data updates prior to the data updates being merged into the data table; and

asynchronously pushing query results of the query to the client device.

2. The method of claim 1 , wherein the plurality of processing stages includes a sorting stage to sort the data updates after ingestion of the data updates by the ingest stage, and wherein the merging stage merges the sorted data updates into the data table.

3. The method of claim 2, and further comprising:

performing a sort process by the sorting stage;

passing the query from the coordinator to the sort process; and executing the query with the sort process on data updates being sorted by the sort process.

4. The method of claim 2, and further comprising:

executing the query between the sort stage and the merging stage.

5. The method of claim 2, wherein the merging stage uses a log-structured merge (LSM) tree for merging, and wherein the method further comprises: executing the query at a first level of the LSM.

6. The method of claim 1 , and further comprising:

storing the results of the query in a queue specified by the client device; and

asynchronously pushing the results of the query as a stream from the queue to the client device.

7. The method of claim 6, wherein the queue uses advanced message queuing protocol (AMQP) to asynchronously push the results of the query to the client device.

8. The method of claim 6, wherein the queue uses web sockets to asynchronously push the results of the query to the client device.

9. The method of claim 1 , and further comprising:

executing the query at a sampling frequency specified by the client device; and

asynchronously pushing results from each execution of the query to the client device.

10. The method of claim 1 , and further comprising:

continuously executing the query; and

asynchronously pushing results from the continuous execution of the query to the client device at a sampling frequency specified by the client device.

11. The method of claim 1 , wherein the results of the query comprise consolidated results collected during a time window specified by the client device.

12. A data processing system comprising: a processing pipeline having a plurality of processing stages, wherein the processing stages perform respective different operations on data updates received by the system, wherein the plurality of processing stages includes an ingest stage to ingest the data updates into the system, and a merging stage to merge the data updates into a data table;

a coordinator to receive a query from a client device and cause the query to be executed on the data updates prior to the data updates being merged into the data table; and

a queue to store query results of the query and asynchronously push the query results to the client device.

13. The system of claim 12, wherein the plurality of processing stages includes a sorting stage to sort the data updates after ingestion of the data updates by the ingest stage, wherein the merging stage merges the sorted data updates into the data table, and wherein the sorting stage executes the query on data updates being sorted by the sorting stage.

14. The system of claim 12, wherein the queue uses one of advanced message queuing protocol (AMQP) and web sockets to push the query results to the client device.

15. A data processing system comprising:

a processing pipeline having a plurality of processing stages, wherein the processing stages perform respective different operations on data updates received by the system, wherein the plurality of processing stages includes an ingest stage to ingest the data updates into the system, a sorting stage to sort the data updates after ingestion of the data updates by the ingest stage, and a merging stage to merge the sorted data updates into a data table;

a coordinator to receive a query from a client device and cause the query to be executed on the data updates prior to the data updates being merged into the data table, wherein the coordinator causes the query to be executed at a frequency specified by the client device; and a queue to store query results of the query and asynchronously push the query results to the client device at a frequency specified by the client device.