Method for distributed RDSMS

• US 8,234,296 B1
• Filed: 11/07/2011
• Issued: 07/31/2012
• Est. Priority Date: 03/15/2003
• Status: Active Grant

• Current Assignee: Guavus, Inc. (Thales SA)
• Original Assignee: SQLstream, Inc. (Thales SA)
• Inventors: Black, Damian
• Primary Examiner(s): Breene, John E
• Assistant Examiner(s): Ly, Anh
• Application Number:
  US13/290,967
• Time in Patent Office: 267 Days
• Field of Search: 707/769, 707/805, 707/954, 707/E17.009, 707/E17.014, 707/E17.129, 707/E17.045, 707/713, 707/770
• US Class Current: 707/769
• CPC Class Codes: G06F 16/24568 Data stream processing; Continuous queries
• CPC Class Codes: G06F 16/83 Querying
• CPC Class Codes: H04L 67/02 based on web technology, e.g. hypertext transfer protocol [HTTP]
• CPC Class Codes: H04L 67/10 in which an application is distributed across nodes in the network software deployment G06F8/60; multiprogramming arrangements G06F9/46
• CPC Class Codes: Y10S 707/922 Communications
• CPC Class Codes: Y10S 707/954 Relational
• CPC Class Codes: Y10S 707/959 Network
• CPC Class Codes: Y10S 707/99931 Database or file accessing
• CPC Class Codes: Y10S 707/99942 Manipulating data structure, e.g. compression, compaction, compilation
• CPC Class Codes: Y10S 707/99943 Generating database or data structure, e.g. via user interface
• CPC Class Codes: Y10S 707/99945 Object-oriented database structure processing

Abstract

Methods for processing streaming data records are provided that include facilities for defining and naming multiple input and output data streams using relational processing definitions and abstractions similar to those found in relational database management systems (RDBMS) and embodied as a set of communicating stream processing nodes. The result is a Distributed Data Stream Management System (DDSMS) which is also a kind of Relational Data Stream Management System (RDSMS). The former comprises a set of Relational Data Stream Management System (RDSMS) nodes, whereas the latter might contain only a single RDSMS node.

Claims

1. A method of managing and processing streams of data records over a data communication network linking a plurality of data processors comprising:

(a) providing a plurality of stream processing nodes for consuming and generating streams of data records executing on said data processors;

(b) providing a central configuration storage for holding current management and configuration data describing the names, relational stream processing operations and definition data of said streams, and for holding configuration and status data of said stream processing nodes, said central configuration storage executing on at least one of said data processors;

(c) configuring said stream processing nodes using a user-interface to browse, create and edit configuration information in said central configuration storage, which in turn communicates over said data communication network with said stream processing nodes to effect configuration changes therein required to maintain consistency of configuration data of said stream processing nodes with the configuration data held in said central configuration storage;

(d) generating with a translator said relational stream processing operations from relational stream processing source definitions input through said user-interface and then storing said relational stream processing operations in said central configuration storage, which in turn communicates with said stream processing nodes to effect configuration changes therein;

(e) communicating with an external data generator to create input streams for processing by said stream processing nodes;

(f) communicating with an external data processor to consume output streams of said stream processing nodes;

(g) transmitting data records in the form of streams comprising sequences of data records each of which comprises a set of named data fields with each field having a defined data type and holding a data value, said transmitting being from some of said stream processing nodes to others of said stream processing nodes, in order to be further processed according to the definitions of the configuration data held in said central configuration storage;

(h) processing streams transmitted to the input of each of said stream processing nodes and creating output streams from said stream processing nodes, said processing being performed within each of said stream processing nodes according to a set of relational stream processing operations by a relational stream engine of each of said stream processing nodes, respectively, and where said relational stream engine has a plug-in management feature to allow dynamic modification and extension of said processing through dynamic loading and unloading executable modules thereto or therefrom;

(i) controlling each of said stream processing nodes through a control interface, thereby remotely controlling execution of each of said stream processing nodes;

(j) configuring said stream processing nodes through a configuration interface whereby current operating parameters, current relational stream processing operations and currently loaded executable plug-in modules of each of said stream processing nodes can be dynamically modified; and

(k) allowing each of said stream processing nodes to interoperate with one another dynamically through said control interface and through said configuration interface in order to control and configure processing of records by each of said stream processing node according to said configuration data stored in said central configuration storage, whereby streams can be retransmitted and redirected from any of said stream processing nodes to any other of said stream processing nodes in order to perform correct processing of records with respect to said configuration data;

whereby information processing systems can be constructed by using said stream processing nodes interfacing with said external data generator and interfacing with said external data processor so as to input external streams of records, process said external streams of records using said stream processing nodes according to the relational stream processing operations stored in the central configuration storage, and output resultant record streams.

2. The method of claim 1 wherein said relational stream processing source definitions are written in a textual language that is a subset of the sql language extended with time window constraints.

3. The method of claim 1 further comprising performing global optimization according to a set of optimization criteria stored in said central configuration storage, said global optimization comprising analyzing the configuration data of said central configuration storage and generating and storing new configuration data in said central configuration storage, and changing the configuration of said stream processing nodes such that the data sent to said external data processor and the data consumed from said external data generator are unaffected but that resources consumed by said stream processing nodes are reduced whereby overall processing performance of said stream processing nodes when operating as a complete system is optimized.

4. The method of claim 1 further comprising performing local optimization within each of said stream processing nodes to improve performance of each of said stream processing nodes.

5. The method of claim 1, further comprising providing recovery processing where said stream processing nodes communicate over the data communication network to other relevant ones of said stream processing nodes to request retransmission of data records in order to recover from detected failures.

6. The method of claim 1 wherein all of said stream processing nodes execute on a single data processing computer.

7. The method of claim 1 wherein said stream processing nodes store streams in a recovery buffer for possible retransmission.

8. The method of claim 7 wherein said recovery buffers are stored in persistent storage.

9. The method of claim 1 wherein the relational stream processing of records is performed in real-time according to timing constraints stored in said central configuration storage.

10. The method of claim 1 wherein said communicating with an external data generator results in generating streams of records each of which are automatically augmented with a timestamp field indicating their time of creation.

11. The method of claim 10 wherein the processing of records is performed in real-time, the timing of said processing being governed by timing constraints stored in said central configuration storage.

12. The method of claim 1 wherein said external data generator generates records in a format produced by external software application components and said external data processor consumes records output by said stream processing nodes, where said application components execute on at least one computer, and wherein said method thereby translates records from said format to required input formats of other of said application components, thereby providing improved translation throughput and system manageability, and real-time matching of records having the same field and values, and clearly separating the data streams and their management from the application components and their management, insulating said application components from change to one another.

13. The method of claim 1 wherein said external data generator generates records representing messages in a format produced by external software application components and said external data processor consumes said records representing messages, wherein said application components execute on at least one computer, and wherein said method thereby brokers messages between said application components, thereby providing improved flexibility, message throughput, system manageability and real-time matching of messages having the same fields and values.

14. The method of claim 1 wherein said external data generator generates records representing streaming market data, and said external data processor passes said records to a plurality of consumers and wherein said method thereby distributes said records as streaming preprocessed market data records to said plurality of consumers, thereby providing improved preprocessing flexibility, system throughput, system manageability and real-time matching of records with common market instrument identity fields.

15. The method of claim 1 wherein said external data generator generates records representing streaming sensor data and said external data processor passes the records to a plurality of consumers, and wherein said method thereby distributes said records as streaming preprocessed sensor data to said plurality of consumers, thereby providing improved preprocessing flexibility, throughput, system manageability and real-time matching of records with common sensor identity fields within a specified time window.

16. The method of claim 1 wherein said external data generator generates records representing streaming location data records for mobile devices and said external data processor passes the records to a plurality of consumers, and wherein said method thereby distributes said records as streaming preprocessed location based data records to said plurality of consumers, thereby providing improved preprocessing flexibility, throughput, system manageability and real-time matching of records with the same location fields within a specified time window.

17. The method of claim 1 wherein said external data generator generates records representing streaming incomplete service detail records and said external data processors passes the records to a plurality of consumers, and wherein said method thereby distributes said records as streaming preprocessed service detail records to said plurality of consumers, thereby providing improved benefits are afforded of superior preprocessing flexibility, throughput and system manageability.

18. The method of claim 1 wherein said external data generator generates records representing streaming incomplete data records and said external data processor passes the records to a plurality of consumers of streaming preprocessed service detail records, and wherein said method thereby merges, aggregates, and otherwise completes said streaming incomplete data records to said plurality of consumers including data warehouse and database systems executing on computers, thereby providing improved preprocessing flexibility, throughput and system manageability.