JP5377897B2 - Stream data ranking query processing method and stream data processing system having ranking query processing mechanism - Google Patents

Abstract

<P>PROBLEM TO BE SOLVED: To (1) achieve a ranking calculating mechanism for maintaining compatibility in inserting stream data into a window and also in extinguishing data, and to (2) achieve a general-purpose interface for delivering a ranking calculation result to an application and an output mechanism according to the interface in order to secure the versatility of a stream data processing system. <P>SOLUTION: (1) A mechanism for managing ranking information using a sign of a stream tuple generated when stream data is inserted into, or deleted from, a window is provided. (2) A mechanism for generating only the differential information of ranking calculation results, a mechanism for adding ranking information according to a request, an interface for generating and outputting all ranking information from the differential information, a mechanism for generating all ranking calculation results, and an interface for using these mechanisms are provided. <P>COPYRIGHT: (C)2009,JPO&amp;INPIT

Description

  The present invention relates to a ranking calculation method and a stream data processing system having the calculation method in a stream data processing system that processes stream data that comes every moment in real time.

  Conventionally, a database management system (hereinafter referred to as DBMS) has been positioned at the center of data management of enterprise information systems. The DBMS stores data to be processed in a storage and realizes highly reliable processing represented by transaction processing for the stored data. On the other hand, there is an increasing demand for a data processing system that processes a large amount of data that arrives every moment in real time. For example, when considering a financial application that supports stock trading, one of the most important issues of the system is how quickly it can react to fluctuations in stock prices. In a system in which stock data is temporarily stored in a storage device as in a conventional DBMS and a search is performed on the stored data, data storage and subsequent search processing cannot keep up with the speed of stock price fluctuations. , You could miss a business opportunity. For example, US Pat. No. 5,495,600 (Patent Document 1) discloses a mechanism in which a stored query is periodically executed. Even in this mechanism, the moment when data enters like the aforementioned stock price. It is important to execute the query. That is, since the difference between the query execution cycle and the data processing timing cannot be tolerated, it has been difficult to apply to the real-time data processing represented by the financial application. The approach of creating various real-time applications individually using a programming language such as Java (R) is a quick response to prolonged development time, rising development costs, and changes in operations that use the applications. However, there is a problem that it is difficult, and a general-purpose real-time data processing mechanism has been demanded.

  A stream data processing system has been proposed as a data processing system suitable for such real-time data processing. For example, R.A. Motwani, J.M. Widom, A. Arasu, B.H. Babcock, S.M. Babu, M.M. Data, G.G. Manku, C.I. Olston, J.M. Rosenstein, and R.R. By Varma: “Query Processing, Resource Management, and Application in a Data Stream Management System”, In Proc. of the 2003 Conf. A stream data processing system STREAM is disclosed in on Innovative Data Systems Research (CIDR) and January 2003 (Non-Patent Document 1).

  In a stream data processing system, unlike a conventional DBMS, a query is first registered in the system, and the query is continuously executed as data arrives. The stream data here is not one large logically continuous data such as a video stream, but stock price distribution data in a financial application, POS data in a retail business, probe car data in a traffic information system, computer system management Is a relatively small amount of time-series data that is logically independent such as an error log and sensing data generated from a ubiquitous device such as a sensor or RFID. Since stream data continues to arrive at the system, processing in real time is impossible if processing is started after waiting for the end of the stream data. In addition, data that arrives at the system must be processed in the order of arrival without being affected by the data processing load. In the STREAM, in order to realize real-time processing of stream data that continues to arrive in the system by specifying a time width such as the latest 10 minutes or a number of widths such as the latest 1000, and cutting out part of the stream data. A concept called a sliding window (hereinafter simply referred to as a window) is introduced. As a suitable example of a query description language including window specification, CQL (Continuous Query Language) disclosed in Non-Patent Document 1 can be cited. CQL is extended by specifying a window by using parentheses following a stream name in the FROM clause of SQL (Structured Query Language) widely used in DBMS. Regarding SQL, C.I. J. et al. Date, Hugh Darwen: “A Guide to SQL Standard (4th Edition)”, Addison-Wesley Professional; 4 edition (November 8, 1996), ISBN: 0201964260 (non-patent document).

  A query 1201 in FIG. 12 is an example of a query by CQL shown in section 2.1 of Non-Patent Document 1. In this query, a domain proxy. The total number of accesses for the past day from the current time from edu is calculated. Requests is Web access data that continues to arrive at the Web proxy server, and is not static data such as a table handled in a conventional DBMS, but is stream data without interruption. Therefore, the total number of accesses cannot be calculated without specifying which part of the stream data is targeted by the window specification “[Range 1 Day Preceding]”. The stream data cut by the window is held in the memory and used for query processing. As a typical window designation method, there are a range window for designating a window width by time and a row window for designating a window width by the number of data. For example, if [Range 10 minutes] is selected using the Range window, the latest 10 minutes are subject to query processing. If [Rows 10] is selected using the Row window, the latest 10 items are subject to query processing. Become.

  Stream data processing systems are expected to be applied to applications that require real-time processing, such as financial applications, retail sales monitoring, traffic information systems, and computer system management. Hereinafter, an application that requires real-time processing is referred to as a real-time application. In real-time applications, it is often necessary to perform ranking calculation at a certain moment in order to quickly extract highly important information from a huge amount of information. For example, in financial applications, ranking information to focus on stock price fluctuations and stocks with large trading volumes is important. In retail sales monitoring, sales and sales figures from various angles, such as by store and by product, are important. Ranking information attracts attention. In addition, traffic information systems require ranking information to focus on areas with high traffic congestion and high traffic volume. In computer management, prioritize management targets such as the number of serious errors and the number of accesses. Ranking information is essential.

  When the ranking calculation target data is static, that is, when the data to be ranked is not changed, the data is sorted by the key for ranking (hereinafter, ranking key), and the ranking of the sorting results According to the above, data may be output. For example, when calculating the top 10 ranking information of sales of stock prices stored in the database, the sales for each brand on that day are aggregated, and the results of aggregation using the sales as a ranking key are sorted. The top 10 items may be selected and output. US Pat. No. 7,251,648 (Patent Document 2) discloses a method for automatically determining a ranking key (sales amount in the above example) from a query input by a user. US published patent US2006 / 0259457 (patent document 3) reduces the cost of extra data processing by adding a condition during query processing when there is a specification to output only the first n rows in a DBMS query A method is disclosed. In the SQL, a GROUP BY phrase for classification by brand, an aggregation function SUM for calculating the total sales amount, and an ORDER BY phrase for performing sorting based on the aggregation value are prepared. By combining these, it is possible to generate ranking calculation results in descending order of sales (or in ascending order) from daily stock transaction data stored in the database.

  However, in the real-time application, since new data (stream data) continues to arrive one after another, it is difficult to make it static. When a ranking calculation for a real-time application is to be performed using a DBMS, it is necessary to store the data in the DBMS every time stream data arrives, and perform the above-described classification, aggregation, and sorting processing by the DBMS. In these processes, it is basically necessary to access a large amount of data in the database, and the processing cost is high. For this reason, when stream data generated from a real-time application arrives at high speed, that is, when the time interval at which stream data arrives is short, it is impossible to execute processing within the time interval, and real-time using DBMS Realizing ranking calculation for applications has been difficult.

  As described above, in the stream data processing system, the processing target is cut out from the infinite stream data and processed in the above-described window. The data to be processed is only the data existing in the window, and the data pushed out from the window needs to be deleted from the target of the ranking process. The timing at which data is pushed out of the window differs depending on whether the window designation method is time (the aforementioned Range window) or the number of cases (the aforementioned Row window). In the case of specifying the number of cases, the time at which the data to be processed is pushed out of the window cannot be determined at the moment when the data enters the window, but is determined by the subsequent stream data. On the other hand, in the case of time designation, the time when the data to be processed is pushed out from the window can be determined at the moment when the data enters the window, but the erasure timing (timing pushed out from the window) Does not synchronize.

  In ranking calculation, it is necessary to maintain ranking information consistency by executing ranking calculation every time stream data is inserted into a window. In addition, it is necessary to maintain the consistency of ranking information when data disappears from the window. In particular, when a window is designated for time, it is necessary to perform ranking calculation in consideration of the disappearance timing of data from a window that is not synchronized with the arrival of subsequent data.

  Furthermore, in the ranking calculation, it is necessary to observe the restrictions of real-time processing, which is one of the important purposes of using the stream data processing system, by improving the processing efficiency. In addition, since the stream data processing system is a general-purpose data processing platform, it responds to the demands of applications to use, such as passing only difference information of ranking calculation results, passing the entire ranking calculation results, and including ranking information in ranking calculation results. It is necessary to prepare a general-purpose interface and a mechanism for realizing processing according to the interface. A ranking calculation method for a stream data processing system that satisfies the above conditions has not been realized so far.

US Pat. No. 5,495,600 US Pat. No. 7,251,648 US Published Patent US2006 / 0259457 JP 2006-338432 A R. Motwani, J.M. Widom, A. Arasu, B.H. Babcock, S.M. Babu, M.M. Data, G.G. Manku, C.I. Olston, J.M. Rosenstein, and R.R. By Varma: “Query Processing, Resource Management, and Application in a Data Stream Management System”, In Proc. of the 2003 Conf. on Innovative Data Systems Research (CIDR), January 2003 C. J. et al. Date, Hugh Darwen: “A Guide to SQL Standard (4th Edition)”, Addison-Wesley Professional; 4 edition (November 8, 1996), ISBN: 0201964260.

  When using a stream data processing system to achieve the ranking calculation required for real-time applications, in addition to inserting stream data into a window, it is possible to realize ranking processing that maintains consistency even when the stream data disappears Necessary. In order to obtain a ranking process result that can be called real time, it is necessary to speed up the ranking update process that is executed every time the internal data of the processing target window changes.

  An object of the present invention is to provide a ranking processing method and system capable of speeding up the ranking update operation that is performed every time the internal data of the processing target window changes, and that maintains the consistency of the processing results. There is.

  Among the inventions disclosed in this application, the outline of typical inventions is as follows. That is, in a typical embodiment, every time stream data is inserted into or deleted from a window, that is, every time a stream tuple starts its lifetime and every time a stream tuple lives, the stream that is in its lifetime Stream data processing that generates and updates those rankings within the range of tuples and stores them in a buffer within the range of stream tuples that are in the lifetime beyond the rank range specified for output is adopted.

  Speaking only from the output of ranking information at a certain point in time, it seems that it is sufficient to store the ranking information if it is performed within the range of stream tuples in the rank specified for output. However, due to acceptance of a new stream tuple, insertion and deletion into the stream tuple in the window occur, and the ranking itself changes each time the insertion and deletion are performed. Therefore, in order to continue to perform consistent ranking calculation, it is necessary to update the ranking information for each insertion / deletion, and the lifetime is beyond the rank range specified for output. It is necessary to store stream tuples within the range of stream tuples and their ranking information.

  Further, the representative embodiment inserts the received stream tuple into a window as a processing target for ranking calculation, determines the lifetime of the stream tuple in the window, and deletes the stream tuple from the window at the end of the lifetime. It has a two-stage processing mechanism of a window manager and a ranking processing module that performs ranking calculation, and the window manager performs not only the information of the entire stream tuple in the window but the window difference information indicating the changing part every moment. The ranking processing module is transmitted to the module, and the ranking processing module is characterized in that the ranking is updated using the transmitted window difference information and saved information that has been calculated and saved in the past. Specifically, the stream tuple to which the code indicating that the stream tuple has been inserted into the window and the stream tuple to which the code indicating that the stream tuple has been deleted from the window are ranked as the window difference information. The ranking processing module updates ranking information based on the difference information, saves it in the ranking information holding buffer, and outputs ranking output information in a designated format.

  By using the present invention, in a stream data processing system that processes a large amount of data that arrives from moment to moment in real time, it is possible to achieve high-speed and high-efficiency ranking calculation while maintaining consistency with input stream data. By applying this ranking calculation method, it is possible to provide a data processing infrastructure that can be commonly used in real-time applications.

  1 and 16 show a preferred implementation example of the stream data processing system of the present invention. A client computer 101 that executes application 1 (102) and a client computer 103 that executes application 2 (104) are connected to a stream data processing system 115 via a network 107. The network 107 may be a local area network (LAN) connected by Ethernet, optical fiber, or the like, or a wide area network (WAN) including the Internet that is slower than the LAN. The client computer may be an arbitrary computer system such as a personal computer or a blade type computer system.

  A computer on which the stream data processing system of this embodiment operates is called a stream data processing server. As shown in FIG. 16, the stream data processing server 1601 is a computer having a communication interface 1602 such as an Ethernet adapter, a CPU 1603, a memory 1604, and an I / O interface 1605, and is an arbitrary one such as a blade computer system or a PC server. Computer system. The stream data processing server accesses the client computer and a data source described later through the communication interface. When the stream data processing server stores the stream data processing result, the intermediate result of the processing, and the setting data necessary for system operation in the nonvolatile storage, the storage device 1606 connected to the stream data processing server can be used. . The storage device 1606 is directly connected via the I / O interface of the stream data processing server, or connected to the network via the communication interface.

  The stream data processing system 115 operates on the stream data processing server 1601. FIG. 1 shows the main components of the stream data processing system. The application first registers a query in the stream data processing system (105). The registered query is stored in the query management table (112). Detailed procedure for query registration, data storage method inside stream data processing system, storage format, analysis method after receiving query, optimization method, system registration method, stream data processing system registration method As for the data holding method in the system, Japanese Patent Application Laid-Open No. 2006-338432 “Query processing method of stream data processing system” (Patent Document 4) discloses a suitable method of implementation. The query management table may be held in the memory 1604 on the stream data processing server or stored in the storage device 1606 connected to the stream data processing server.

  A large amount of data comes to the stream data processing system from the stream data source 1 (122) to the stream data source N (123), which are one or more stream data sources, via the network 121 every moment. This data is called stream data. Preferable examples of stream data include stock price distribution information in a financial application, POS data in a retail business, probe car information in a traffic information system, an error log in computer system management, and the like. In the stream data processing system, the stream data received by the data flow manager 119 is fed to the query processing engine 113 via the communication interface 1602.

  As described above, a window is used in a stream data processing system in order to handle stream data that is a relatively small amount of logically independent time series data that continuously arrives. The window manager 126 in FIG. 1 generates a stream tuple by applying the window operation specified by the query to the incoming stream data, and sets the lifetime of the stream tuple in the system. The time when the stream data is inserted into the window corresponds to the start time of the lifetime, and the time when the stream data is deleted from the window corresponds to the end time of the lifetime.

  The configuration and operation contents of the window manager will be described with reference to FIG. The window manager 126 includes a stream data reception interface 1505, a stream tuple holding buffer 1502, a lifetime determination unit 1504, and a difference information generation unit 1506. The stream data reception interface stores that the stream tuple as a component of the received stream data is stored in the stream tuple holding buffer 1502 and stored in the lifetime determination unit 1506. The lifetime determination unit 1506 determines the lifetime of each stream tuple by the window operation, and erases the stream pull whose lifetime has ended from the stream storage buffer 1502. The difference information generation unit 1504 generates a plus tuple at the timing when the stream tuple is stored in the stream tuple holding buffer, and outputs it as difference information (1508). Similarly, a minus tuple is generated at the timing at which the stream tuple is deleted from the stream tuple holding buffer (the timing at which the lifetime of the stream tuple ends), and similarly output as difference information (1508).

  As shown in FIG. 16, the stream tuple holding buffer 1502 is arranged on a memory allocated by the memory manager 117 in the query processing engine 113. The memory is connected to the memory 1604 on the stream data processing server 1601 or the storage device 1606 connected to the stream data processing server or the stream data processing server via a network according to the required performance requirement and reliability requirement. Alternatively, a memory on a server computer (block 133 in FIG. 1) that holds the same computer resources as the stream data processing server may be used.

  Next, the configuration of the ranking processing module will be described with reference to FIG. The ranking processing module 116 includes a stream tuple reception interface 206, an order / rank generation unit 204, a ranking information holding buffer 202, a ranking information holding table 203, a ranking management index 208, and a ranking processing result output interface 207. As shown in FIG. 16, the ranking processing module is arranged on a memory allocated by the memory manager 117 in the query processing engine 113. The memory is connected to the memory 1604 on the stream data processing server 1601 or the storage device 1606 connected to the stream data processing server or the stream data processing server via a network according to the required performance requirement and reliability requirement. Alternatively, a memory on a server computer (block 133 in FIG. 1) that holds the same computer resources as the stream data processing server may be used.

  FIG. 17 shows a ranking information generation sequence in the stream data processing of the present invention. The window manager 126 receives stream data coming from an external data source (1701) and generates difference information (signed stream tuple) (1702). In the ranking processing module 116, the stream tuple reception interface 206 receives the difference information (signed stream tuple) output from the window manager and transfers it to the order / rank generation unit 204. The order / rank generation unit adds the received signed stream tuple to an appropriate position in the ranking buffer while referring to the ranking information holding buffer 202 or deletes the ranking information at an appropriate position in the ranking buffer. A difference (ranking difference information) between the ranking information and the output time is generated (1705), and the ranking difference information is transferred to the ranking processing result output interface 207 (1706). The ranking processing result output interface outputs ranking information in accordance with the data output format specified by the query.

  Hereinafter, the ranking calculation method of the present embodiment for specific queries and stream data will be described. A query 301 shown in FIG. 3A is a query that commands a ranking process that does not specify rank output (outputting the rank itself). The SELECT clause on the first line is s. id and s. The output value is a set of val values, the FROM phrase in the second line is the stream s, and the Partition By phrase in the second line is the s. By grouping with the value of id, each s. Keeping the latest one of the values of val, and the LIMIT phrase on the third line is s. This means that 3 items are output in descending order of the value of val. In the stream data processing system of the present embodiment, in a query inputted in advance, the columns to be subjected to ranking calculation, the ranking direction (ascending / descending order), the number of ranking calculation results to be output, and the calculation results It is specified whether or not rank information is added. In query 301, the LIMIT phrase on the third line is s. It is a ranking specification that outputs three items in descending order of the value of val. Further, the IDSTREAM phrase on the first line means that this query outputs only the difference information from the previous output as the output.

  With reference to FIG. 4 and FIG. 5A, processing contents when stream data arrives for the stream data processing system in which the query 301 is set will be described. When the query 301 is registered in the stream data processing system 115, for example, query analysis, query optimization, and query generation are executed by the method disclosed in Patent Document 4, and the query execution format is registered in the query processing engine. Since the query 301 includes a ranking calculation specification, the ranking calculation module 116 in the query processing engine 113 executes the ranking calculation when executing the execution format query. As described in Patent Document 1, in a stream data processing system, a query continues to operate on the system after being registered, and the state of each query changes each time stream data arrives at the system.

  Assume that the stream data shown in FIG. 4 has arrived in a situation where the query 301 is registered. 4 and 5 (a) schematically show how each processing module processes stream data that has arrived at the system as time passes, with the passage of time taken along the vertical axis. As shown in the legends 402 and 502, in this embodiment, the incoming stream data is assumed to be in the form of {id, val}, which is represented by an ellipse. t1 to t7 (401) represent times when the stream data 405 to 411 arrive at the stream data processing system. For example, stream data {a, 50} (405) and stream data {b, 10} (406) indicate that they arrive at the stream data processing system at times t1 and t2, respectively. The horizontal axis of FIG. 4 represents the position where the incoming stream data is processed and the data to be generated. Further, the rounded square (1502) on the left side of FIG. 4 performs a window operation “[Partition By s.id Rows 1]” (403) by the window manager (126) on the stream data that has arrived at the system. The state at each time of the stream holding buffer storing the applied result is shown. The right rounded square (202) is a ranking calculation specification phrase “LIMIT 3 By in the order / rank generation unit (204) in the ranking processing module (116) for the stream tuple coming from the window manager. The status of the ranking information holding buffer storing the application result of “s.val DESC” (404) at each time is shown.

  As described above, the window manager 126 generates a stream tuple by applying the window operation specified by the query to the incoming stream data, and sets the lifetime of the stream tuple in the system. In the example of FIG. 4, the start time of the survival period is represented by a black circle (426) and the end time is represented by a white circle 427, and the survival time of the stream tuple {a, 50} (405) is obtained at time t1 by the window operation. It starts and ends at time t7. In this implementation example, at the start time of the lifetime of the stream data, a tuple (hereinafter referred to as a plus tuple) in which a sign indicating an increase is added to the stream data is generated inside the system. Further, when stream data is deleted from the window, a tuple (hereinafter referred to as a minus tuple) with a reference representing a decrease and having a reference to the previously output plus tuple is generated. The above processing is executed by the window manager. For example, in the case of FIG. 4, a plus tuple 430 corresponding to the stream data {a, 50} (405) is generated at time t1, and a minus tuple 431 of the stream data is generated at time t7. Subsequent query processing subsequent to the window operation is executed with respect to difference information generated due to the plus tuple and the minus tuple at the arrival timing of the main tuple and the minus tuple. The concept of plus tuple and minus tuple itself is introduced in Non-Patent Document 1 described above.

  In the query 301, the incoming stream data is first s.d. by the window operation “[Partition By s. Id Rows 1]” (403). Only the latest one is held for each id value. Specifically, in FIG. 4, stream data {a, 50} (405) at time t1, {b, 10} (406) at time t2, {c, 30} (407) at time t3, time {d, 20} (408) arrives at t4, and {e, 40} (409) arrives at time t5. These five stream data are s. Since the value of id is different, it is held in each window. Next, when new stream data {e, 15} (410) arrives at time t6, s. The stream data 409 whose id value is e is pushed out (deleted) from the window. Next, when {a, 45} (411) arrives at time t7, {a, 50} (405) is similarly deleted from the window. This is illustrated in the central portion 429 of FIG. For example, stream data {a, 50} (412) arrives at time t1, and is held in the window until {a, 45} (418) arrives at time t7. At this time, the lifetime of the stream data {a, 50} is expressed as t1 to t7 (however, the time t7 is not included). In the example of FIG. 4, the start time of the survival period is indicated by a black circle (426), the end time is indicated by a white circle (427), and the lifetime is indicated by a solid line. Similarly, the lifetime of {e, 40} (417) is from t5 to t6. On the other hand, for {b, 10} (413), {c, 30} (414), {d, 20} (415), {e, 15} (417), and {a, 45} (418), Since the end time is not determined at time t7, it is represented by a solid line.

  As described above, how ranking information is generated from stream data having a lifetime will be described with reference to FIG. The right side 432 of FIG. 4 shows the processing result by the ranking calculation designation phrase “LIMIT 3 By s.val DESC” (404). As described above, the meaning of this ranking specification phrase is s. It is to extract 3 cases in descending order by the value of val. Now, stream data {a, 50} (419), {b, 10} (420), {c, 30} (421) arrive at times t1, t2, and t3, respectively. These three stream data are respectively output as the top three ranking information. Next, when {d, 20} (422) arrives at time t4, the s. Since the value 20 of val is larger than 10 of {b, 10} held so far, {b, 10} is deleted from the ranking information of the top three cases, and its lifetime ends at t4 ( 428), {d, 20} is output as a ranking calculation result. Next, when {e, 40} (423) arrives at time t5, the s. Since the value of val is 40, it is included in the top three cases, so it is output as ranking information. Then, {d, 20} having the smallest value among the top three held so far is deleted from the ranking. However, at time t6, as described above, {e, 40} (416) is deleted from the window due to the arrival of {e, 15} (417), so {d, 20} (424) is again the top 3 Will be revived and will be output again as a ranking calculation result. At time t7, {a, 50} (412) is replaced with {a, 45} (418), and s. Since the val value 45 is included in the top three cases, {a, 45} (425) is output as the ranking result.

  A preferred implementation method for the above ranking calculation will be described with reference to FIGS. 2, 9, 10, 18, and 19. The stream tuple reception interface 206 of the ranking processing module 116 receives a stream tuple as difference information (902). Then, the order / rank generation unit 204 executes buffer maintenance of the ranking information holding buffer 202 (903). A processing method of the buffer maintenance process will be described with reference to FIGS. When receiving the stream tuple, the order / rank generation unit 204 checks whether the sign of the stream tuple is positive or negative (1002). When the sign is positive (when Yes is selected in 1002), the received stream tuple is added to the ranking information holding table 203 of the ranking information holding buffer 202 (1003), and the ranking information holding buffer maintenance process is terminated. Details of the stream tuple addition processing to the ranking information holding buffer will be described with reference to the flowchart of FIG. When receiving the stream tuple to be added (stream tuple with a plus sign), the rank / order generation unit checks whether the rank management index 208 is added to the rank information holding table 203 of the addition destination (1802). The rank management index may be a B + tree index or a hash index using the column to be ranked as a key. When the rank management index exists (when Yes is selected in step 1802), the rank / order generation unit determines the insertion position of the stream tuple to be added to the ranking information holding table using the index. (1803). If the index does not exist (No is selected in step 1802), the rank / order generation unit searches the ranking information holding table, and determines the insertion position of the stream tuple to be added to the ranking information holding table. Determine (1804). After determining the insertion position, the rank / order generation unit inserts the stream tuple to be added into the ranking information holding table (1805), and if a rank management index exists (if Yes is selected in step 1806). The index is also updated, and the stream tuple addition processing to the ranking information holding buffer is completed (1808).

  In a preferred embodiment of the ranking information holding buffer, the order relation of the added stream tuples in the column value designated for ranking is held. The reason for maintaining the order relationship when adding a stream tuple is that it is difficult to output the immediate result, which is a request of the real-time processing application, in the method of creating the order relationship collectively by delay processing. For example, in the case of the query 301 in FIG. 3, the column to be ordered (the above-described ranking key) is s. In the ranking information holding table (203) in the ranking information holding buffer of FIG. The rank information and stream tuple are held in descending order of val.

  Returning to FIG. 10, when the sign of the received stream tuple is negative (when No is selected in 1002 of FIG. 10), the tuple with positive sign corresponding to the stream tuple from the ranking information holding buffer (202). Is deleted (1004), and the ranking information holding buffer maintenance process is terminated.

  Details of the stream tuple deletion processing from the ranking information holding buffer will be described with reference to the flowchart of FIG. The rank / order generation unit checks whether or not the rank management index 208 is assigned to the ranking information holding table 203 (1902). When the rank management index exists (when Yes is selected in step 1902), the rank / order generation unit uses the index to search for a stream tuple to be deleted in the ranking information holding table (1903). ). If the index does not exist (No is selected in step 1902), the rank / order generation unit searches the ranking information holding table and determines a stream tuple to be deleted (1904). When the stream tuple to be deleted is determined, the rank / order generation unit executes the deletion process of the stream tuple (1905). If the rank management index exists (if Yes is selected in step 1906), the rank management index is also updated, and the stream tuple deletion process from the ranking information holding buffer is terminated.

  Here, with reference to FIG. 14, the ranking held in the ranking information holding table when the stream arrives in the time chart shown in FIG. 4 for the stream data processing system in which the query of FIG. 3A is registered. Explain how information changes. In FIG. 14, t0 (1416), t1 (1404),..., T7 on the left is the time, and the center signed ellipse (1405, 1407,...) Is the difference information (signed stream tuple) generated by the window manager. ) And the right table (1417, 1406,..., 1415) represent the ranking information held in the ranking information holding table. For convenience, it is assumed that no ranking information exists at t0 before time t1.

  When a stream tuple {a, 50} (1405) having a plus sign arrives at time t1 (1404), the stream tuple is registered in the ranking information holding table (1406). Next, when a stream tuple {b, 10} (1407) having a plus sign arrives at time t2 (1418), its ranking object s. The value 10 of val is set to s. of the stream tuple held in the ranking information holding table. Compared with the value 50 of val, the insertion position is determined to be next to {a, 50} (1405), and the stream tuple {b, 10} is inserted at the insertion position (1408). Similarly, when {c, 30}, {d, 20}, {e, 40} arrive at t3, t4, and t5, respectively, these stream tuples are assigned to the s. Sorted and registered in the order of the value of val.

  As shown in FIG. 4, when {e, 15} (408) arrives at time t6, {e, 40} is deleted from the window in the one-row designated Row window query shown in FIG. 433). Therefore, the stream tuple reception interface 206 of the ranking processing module (116) shown in FIG. 14 includes a stream tuple {e, 40} (1413) with a minus sign and a stream tuple {e, 15} with a plus sign. (1414) arrives. The order / rank generation unit 204 searches and determines a stream tuple with a plus sign corresponding to the stream tuple with a minus sign {e, 40} (second from the top of 1412), and deletes the stream tuple. . Then, the insertion position of a new stream tuple {e, 15} (1414) with a plus sign is determined (fourth from the top of 1415), and the stream tuple is added to the ranking information holding table. The same applies to time t7.

  In this embodiment, the case where the data structure in the ranking information holding buffer is a table is shown. However, as another preferable implementation example of the data structure in the ranking information holding buffer, for example, a ranking as shown in FIG. A binary search tree with a key as a node can be cited. In FIG. 13, a node (ranking key) of the binary search tree is indicated by a square 1301, and a data body (stream tuple) pointed to by the node is indicated by a circle and ellipse set 1302. When using a binary search tree, with the ranking key as a node, the value of the ranking key of the left child of all nodes and all of its descendant nodes is smaller than the value of the ranking key of the node, and the right child and all of them. The value of the descendant node is configured to be equal to or greater than the ranking key value of the node. The binary search tree in the rounded circle 1303 in FIG. 13 is an example of the configuration of the binary search tree of the data held in the ranking information holding buffer at the time t5 in FIG. 4, and the contents of the held data Is equivalent to the ranking information holding table 1412 of FIG. By managing the order based on the ranking key using the binary search tree, the management of the order relation when the plus tuple and the minus tuple arrive can be made efficient. Further, as another suitable data structure for maintaining and managing the order relation of stream tuples, B + -Tree used in many DBMS data management mechanisms may be used.

  In the ranking information holding buffer, it is necessary to hold not only the number of ranking outputs specified by the user but also all stream tuples corresponding to the stream data during the lifetime. For example, in the query 301 of FIG. 3, the user is instructed to output only the top three items, but when the stream data series shown in FIG. 4 arrives in the system, at time t4, {d, 20} When (408) arrives, the stream tuple 433 corresponding to the stream data {b, 10} (406) having the fourth rank should not be deleted from the ranking information holding buffer. The reason is that in stream data processing, the ranking changes when the new stream data arrives in the system and also when the lifetime of the stream data ends, so it is currently outside the range specified by the user. This is because the stream data needs to be output again as a ranking result when the lifetime of the other stream data ends. For example, in FIG. 4, the stream data {d, 20} (408) included in the top three cases arrives at time t4 and is pushed out of the ranking by {e, 40} (409) arrived at time t5. However, {e, 40} is deleted from the window by {e, 15} (410) that arrived at time t6 (433), so {d, 20} is included in the ranking calculation result again (424) )There is a need. That is, in the ranking information holding buffer, not only the number of ranking output specified by the user, but also all the stream tuples corresponding to the stream data managed in the window, that is, all the stream tuples corresponding to the live stream data are held. There is a need to.

  However, if the stream tuple is not included in the top 50 at the moment of arrival, the stream tuple is not subject to ranking. It is possible to change the number of stream tuples held in the ranking information holding buffer.

  Returning to FIG. 9, the order / rank generation unit checks whether the processing result affects the ranking based on the result of the buffer maintenance processing (903) of the received stream tuple (904). The case where the processing result affects the ranking refers to the case where the order of the range specified by the query is changed as a result of the maintenance of the ranking information holding buffer based on the received stream tuple. For example, in the query 301 of FIG. 3, since the top three cases are specified as the output range, if the order within the top three is changed, it is determined whether the processing result affects the ranking (step 904). ) Becomes Yes. For example, in the case of the process shown in FIG. 4, when {d, 20} arrives at time t4, the order within the top three is changed, so this is an example of Yes. If the processing result affects the ranking (if Yes is selected in step 904), it is checked whether there is an order output designation (905). If there is an order output designation (if Yes is selected in step 905), a rank information column is added to the processing result tuple (906), the processing result tuple is output (907), and the ranking process is terminated ( 908). If there is no designation of rank output (when No is selected in step 905), a processing result tuple is output (907), and the ranking process is terminated (908).

  In the case of the query 301 of FIG. 3, since the rank output is not specified, the processing result tuple output from the ranking processing result output interface is, for example, shown at 523 in FIG. 5, and no rank information is added. An example of the case where the rank output is designated (when Yes is selected in step 905) will be described later.

  As described above, in ranking calculation of a real-time application, it is necessary to instantly extract useful information from an enormous amount of information. Therefore, it is necessary to improve processing efficiency. Therefore, the stream data processing system according to the present embodiment includes an interface that outputs only the difference in ranking, and an interface that outputs all data in the ranking at that time. The center 202 in FIG. 5A is the window operation “[Partition By s. Id Rows 1]” of the query 301 (FIG. 3) described in FIG. 4 and the ranking calculation designation phrase “LIMIT 3 By s.val DESC”. This shows the status of the ranking information holding buffer storing the processing results at each time. In the query 301, the IDSTREAM phrase is specified as the final output format. The IDSTREAM phrase is an interface that outputs a tuple added to the ranking as an increase tuple and a tuple deleted from the ranking as a decrease tuple as a result of ranking calculation. In FIG. 5A, the stream data after being processed by the IDSTREAM phrase (504) is shown in a rounded square 523 on the right side of the figure. However, the black circle of the stream data represents an increase, and the white circle represents a decrease. The contents of the IDSTREAM phrase processing will be described below. Since {a, 50} (505) is added to the ranking at time t1, {a, 50} (512) is output as the stream data corresponding to the increased processing result in the IDSTREAM phrase. Next, {b, 10} (506) and {c, 30} (507) are added to the rankings at times t2 and t3, respectively, so {b, 10} (514) and {c, 30} (515) Is output as an increment. At time t4, {d, 20} (508) is inserted into the ranking and {b, 10} (506) is deleted from the ranking. In this case, in the IDSTREAM phrase, {d, 20} (516) is output as an increase and {b, 10} (517) is output as a decrease at time t4. By calculating only the information on the increase and the decrease and transmitting it to the client computer that uses the information, the processing cost and the communication cost of the stream data processing system can be reduced. For example, in the case of FIG. 5A, eleven processing results are transmitted to the client computer between t1 and t7. However, when all the top n results are transmitted at each timing, n It is necessary to transmit 7 processing results, and the effect of the present invention in which only difference information is transmitted is particularly high when n is large.

  However, if it is necessary for the client computer side to continue to provide the ranking information to the user at any time, it is necessary to continue to create the overall ranking information from the difference information on the client computer side. In this processing, since it is necessary to manage the state on the client computer side, it may be difficult to realize depending on the state of the client computer and the form of the real-time application. In order to make it possible to use ranking information even in such a situation, the stream data processing system of the present application includes an interface that generates and outputs overall ranking information from the generated difference ranking information. The query 302 in FIG. 3B is the same as the query 301 except for the RSTREAM phrase in the first line, and the SELECT phrase in the first line is s. id and s. The output value is a set of val values, the FROM phrase in the second line is the stream s, and the Partition By phrase in the second line is the s. By grouping with the value of id, each s. The LIMIT phrase in the third line is s. Specifies that 3 items are output in descending order of the value of val. The IDSTREAM phrase on the first line of the query 301 outputs only the difference information from the previous output, whereas the RSTREAM phrase on the first line of the query 302 has an output specification range within each output timing. This means that ranking information for all stream tuples is output. FIG. 5B shows a ranking calculation output result at the interface. Each time state of the ranking information holding buffer (202) in FIG. 5 (b) is the same as in FIG. 5 (a). After the processing of the LIMIT phrase, the result of applying the RSTREAM phrase of the query 302 shown in FIG. 3B is the stream data shown in the rounded square 526 on the right side of FIG. 5B. Hereinafter, the output format of the ranking calculation result will be described with reference to FIG. Since {a, 50} (505) is added to the ranking at time t1, {a, 50} (527) is output in the RSTREAM phrase. Next, when {b, 10} (506) is added to the ranking at time t2, the RSTREAM phrase outputs the entire ranking, that is, two stream data of {a, 50} and {b, 10} (528). . Next, when {c, 30} (507) is added to the ranking at time t3, {c, 30} is output in addition to {a, 50} and {b, 10} (529). Next, {d, 20} (508) is inserted into the ranking at time t4, and {b, 10} (506) is deleted from the ranking. In this case, in the RSTREAM phrase, {a, 50}, {c, 30}, {d, 20} are output at time t4 (530). In order to realize this processing, when the processing result is output, the ranking processing result output interface (207) holds the output content at the previous output, and the output content and the order / rank generation unit (204) newly generate it. The output information may be generated by combining the ranking information. In this explanation, the output timing of the RSTREAM phrase is the moment when the input stream data arrives. However, in order to reduce the load of output generation and communication cost, for example, at regular intervals such as every 1 second, n input tuples For example, every m output tuples can be used.

  Next, an example in the case where the rank output is specified in the query (when Yes is selected in step 905 in FIG. 9) will be described using queries 601 and 602 in FIG. In the query 601, the s. id and s. In addition to val, the rank information at that time is specified to be output by the RANKING AS rank specification on the first line. Further, only the difference information after the query 601 is output last time is output, whereas the query 602 outputs all the ranking information for each output timing.

  First, a ranking calculation method in the case of including rank output designation will be described with reference to FIG. The rounded square (1502) on the left side of FIG. 7 shows the state at each time of the stream tuple holding buffer storing the processing result of the window operation “[Partition By s. Id Rows 1]”. The same. A ranking calculation method including rank output designation for each time state will be described. As shown in the legend 702 in FIG. 7, a set of two values surrounded by an ellipse represents stream data in the format of {id, val}. Also, as shown in the legend 703, a set of three values surrounded by a rounded rectangle represents stream data in the format {rank, id, val}. Here, rank is a rank based on the value of val at the time of output.

  When {a, 50} (705) arrives at the time t1 in the stream data, this stream tuple is included in the ranking and its rank is first, so the ranking calculation result is {1, a, 50} (712 ) Next, when stream data {b, 10} (706) arrives at time t2, since this stream data is also included in the ranking, {2, b, 10} (713) is output. When {c, 30} (707) arrives at time t3, this stream data is also included in the ranking, and the value 30 of val is larger than {b, 10}. 10} is ranked 3rd. Therefore, {2, b, 10} is deleted from the ranking calculation result, and {3, b, 10} (714) and {2, c, 30} (715) are added. Subsequently, when the stream data {d, 20} (708) arrives at time t4, the val value 20 is larger than the val value 10 of {b, 10}, so {3, b, 10} is ranked. It is deleted from the calculation result, and {3, d, 20} (716) is added to the calculation result. Next, when stream data {e, 40} (709) arrives at time t5, its val value 40 is larger than {c, 30} and {d, 20} included in the current ranking calculation result, Since the ranking is second, {2, e, 40} (718) is included in the ranking calculation result. At the same time, {3, d, 20} is deleted from the ranking calculation result, and the order of {c, 30} changes from 2nd to 3rd, so {2, c, 30} is deleted, {3, c, 30} (717) is added. The same applies when {e, 15} (710) and {a, 45} (711) arrive at times t6 and t7, respectively. These ranking calculation processes are executed by the order / rank generation unit (204) of FIG. 2, and the ranking calculation process results at each time are stored in the ranking information holding buffer (202). The preferred implementation method of the calculation is common to the ranking information holding buffer maintenance (903) in the flowchart of FIG. Next, it is checked whether or not there is an order output designation (905). If there is an order output designation (when Yes is selected in 905), a rank information column addition process for the processing result tuple is performed (906). . The ranking information addition process uses the ranking information processing buffer (202). As described above, in the preferred embodiment, the ranking information holding buffer manages the added stream tuple while holding the order relation of the column values for which the ordering is specified. For example, in the case of the query 601, as in the case of the query 301, the column to be ordered is s. Since it is val, the data structure in the ranking information holding buffer includes a table format as indicated by 203 in FIG. A binary search tree using the value of val as a key can be cited as an implementation method. In the rank information addition process, the stream column to which rank information is added is s. When val is used as a key, the position number is calculated, and the rank is added to the column position specified by the query. For example, in the case of the query 601, since “RANKING AS rank” is specified in the first column of the SELECT clause, the rank information is output in the first column.

  In the query 601, the IDSTREAM phrase is specified as the final output format. As described using the query 301, the IDSTREAM phrase is an interface that outputs the stream data added to the ranking as an increase and the stream data deleted from the ranking as a decrease as a result of the ranking calculation. In FIG. 8A, the stream data after being processed by the IDSTREAM phrase (804) is shown on the right side (819) of the figure. When {a, 50} (805) arrives at time t1, the rank information calculated by the order / rank generation unit is added to the head of the output column and output in the first column specified by the query. When {a, 50} arrives, there is only one stream data and the rank is first, so {1, a, 50} (811) is output. Next, when {b, 10} (806) arrives at time t2, the rank becomes second, so {2, b, 10} (812) is output. When {c, 30} (807) arrives at time t3, the val value 30 is smaller than the val value 50 of {a, 50} and larger than the val value 10 of {b, 10}. The ranking is 2nd. Therefore, at time t3, {3, b, 10} (813) and {2, c, 30} (814) are output as increments, and {2, b, 10} (815) as increments. Is output. In the example of the query 301 shown in FIG. 5, since the set itself included in the ranking output did not change due to {c, 30}, only {c, 30} was output as an increase (515). In the example of the query 601, since the order changes with the arrival of {c, 30}, three stream data are output by adding the increment and decrement. Similarly, at times t5 and t6, four (820) and (821) respectively correspond to the change in rank, and the number of stream tuples to be output is increased compared to the case of the query 301. In the query 601, since the IDSTREAM phrase is specified as the final output format, stream data corresponding to the increase / decrease is output. There is also a request to output the entire ranking of each moment. An example is the query 602. By specifying RSTREAM instead of IDSTREAM of the query 601 (829), the entire ranking calculation result at the time of output is output. FIG. 8B shows the execution result of the query 602. Respective outputs at times t1 to t7 are shown in 822 to 828.

In the above embodiment, the designation of ranking is in descending order and only the first ranking is shown, but the designation of ranking may be in ascending order. The ranking start order can be specified by an arbitrary integer value. For example, the query 1101 in FIG. id and s. The set of val values is denoted by s. By grouping with the value of id, each s. The latest one of the values of val is held, and s. It shows that 3 items are output in ascending order of the value of val. The difference from the query 301 in FIG. 3 is the OFFSET phrase and ASC designation on the third line, where the former is the designation of the start order and the latter is the designation in ascending order. If there is an OFFSET designation, a check (904) on whether the received tuple in the ranking process in FIG. 9 affects the ranking is performed by checking whether the received stream tuple is a LIMIT phrase from the start order designated by the OFFSET phrase. It is sufficient to check whether or not the output specification range specified in is affected. Retention and management of ranking information can be processed in the same manner as when OFFSET is not designated in the ranking processing module (116) shown in FIG.
Further, in the above-described embodiment, the ranking target column is explicitly input to the system, but the present invention can also be applied to a system that automatically determines the ranking target column.

The figure which shows the structure of the stream data processing system in this invention. The figure which shows the structure of the ranking process module in this invention. An example of a query including ranking designation (not including rank output designation) in the present invention. The figure which shows the calculation content (rank output designation | designated is not included) of the ranking calculation in this invention. The figure which shows the output result (The ranking output designation is not included) of the ranking calculation result in this invention. The example of the query (rank output specification is included) containing the ranking specification in this invention. The figure which shows the calculation content (rank output designation | designated) of the ranking calculation in this invention. The figure which shows the output result (an order | rank output specification is included) of the ranking calculation result in this invention. The flowchart which shows the ranking process sequence in this invention. The flowchart which shows the ranking information holding buffer maintenance processing procedure in this invention. An example of a query (including offset designation) including ranking designation in the present invention. An example of query description in the query processing language CQL. The example of the expression by the binary search tree of the ranking information in this invention. The figure which shows the mode of a change of the ranking information holding table in the ranking information holding buffer in this invention. The figure which shows the structure of the window manager in this invention. The figure which shows the implementation example on the computer of the query processing engine in this invention. The sequence diagram which shows the ranking calculation method in this invention. The flowchart which shows the stream tuple addition processing method to the ranking information holding buffer in this invention. The flowchart which shows the stream tuple deletion processing method from the ranking information holding buffer in this invention.

Explanation of symbols

115: Stream data processing system 107, 121 ... Network 113 ... Query processing engine 116 ... Ranking processing module 202 ... Ranking information holding buffer 203 ... Ranking information holding table 204 ... Order / rank generation unit 206 ... Stream tuple reception interface 207 ... Ranking Processing result output interface 208 ... ranking management index 116 ... ranking processing module 117 ... memory manager 126 ... window manager 1502 ... stream tuple holding buffer 1504 ... lifetime determination unit 1505 ... stream data reception interface 1506 ... difference information generation units 301 and 302, 601, 602, 1101, 1201 ... Query 1601 ... Stream data processing server 1602 ... Communication interface 603 ... CPU
1604 ... Memory 1605 ... I / O interface 1606 ... Storage device

Claims (11)

In a stream data processing system having a control unit that continuously receives stream data composed of a plurality of stream tuples with time stamps and continuously executes query processing on the stream data by a pre-registered query , Stream data ranking query processing method,
The controller is
In accordance with the window specification specified by the query, for each timing when each stream tuple arrives , the lifetime in the stream tuple window or the end of the lifetime in the stream tuple window that arrived in the past is determined. Generating window difference information indicating insertion of a stream tuple into the window and deletion of the stream tuple from the window;
In accordance with the ranking process specified by the query , the ranking information indicating the ranking between the stream tuples within the range of the stream tuple in the lifetime in the window based on the window difference information is changed from the first ranking information to the second ranking information. Updated to the ranking information of
Generating ranking difference information that is a difference between the first ranking information and the second ranking information;
A ranking query processing method for stream data, comprising: outputting a ranking processing result based on the ranking difference information and an output designation range designated by the query.   2. The stream data ranking query processing method according to claim 1, wherein the control unit stores the updated ranking information of stream tuples that are within the lifetime beyond the specified output range.   The stream data according to claim 1, wherein the control unit outputs a difference between the first ranking information and the second ranking information as the ranking processing result based on the ranking difference information. Ranking query processing method.   The control unit stores the outputted ranking process result, and outputs the entire ranking information included in the output designated range as the ranking process result based on the ranking difference information and the ranking process result outputted last time. The stream data ranking query processing method according to claim 1,   2. The stream data ranking query processing method according to claim 1, wherein the ranking information is stored over all stream tuples within the lifetime.   2. The ranking query processing method for stream data according to claim 1, wherein the ranking information is stored in the window as a table in which each stream tuple is associated with a rank.   A stream data processing system that continuously receives stream data composed of a plurality of stream tuples with time stamps attached thereto, and continuously executes query processing on the stream data according to a query registered in advance.
  A window manager that performs the window operation specified in the query on the incoming stream tuple and determines the lifetime in the window of each stream tuple;
  A ranking processing module that performs ranking processing specified by the query and outputs a set of stream tuples included in the output specification range specified by the query;
  The window manager
  Generates window difference information indicating insertion of a stream tuple into the window and deletion of a stream tuple from the window at each timing when each stream tuple arrives, and transmits it to the ranking processing module.
  The ranking processing module, based on the window difference information transmitted from the difference information generation unit of the window manager, converts the ranking information into first ranking information within the range of the stream tuple in the lifetime in the window. To the second ranking information from
  Storing the second ranking information in the ranking information holding buffer within the range of the stream tuple in the lifetime in the window;
  Generating ranking difference information that is a difference between the first ranking information and the second ranking information, and outputting a ranking processing result based on the ranking difference information and an output designation range designated by the query; A stream data processing system.   8. The ranking processing module according to claim 7, wherein the ranking processing module outputs a difference between the first ranking information and the second ranking information as the ranking processing result based on the ranking difference information. Stream data processing system.   The ranking processing module stores the output ranking processing result, and based on the ranking difference information and the ranking processing result output last time, the ranking information as a whole is included in the output designation range as the ranking processing result. The stream data processing system according to claim 7, wherein the stream data processing system outputs the stream data.   8. The stream data processing system according to claim 7, wherein the ranking information holding buffer holds the ranking information as a table in which each stream tuple and its rank are associated with each other.   When the stream tuple arrives, the difference information generation unit adds a code indicating insertion into the window to the stream tuple, and the stream tuple from the window at the end of the determined lifetime of the stream tuple. 8. The stream data processing system according to claim 7, wherein another code indicating deletion is added, and the difference information is transmitted as the window difference information to the ranking processing module.

Images