Abstract:
System, method, computer program product embodiments and combinations and sub-combinations thereof for moving a project in a complex event processing cluster are provided. In an embodiment, project movement in a complex event processing cluster initiates once project persistence capability exists. A handoff of the project from a source node to a destination node in the complex event processing cluster is coordinated by a manager without losing messages and without relying on a message guarantee delivery protocol.

Description:
BACKGROUND 
       [0001]    1. Field of the Invention 
         [0002]    The present invention relates generally to data computing environments and, more particularly, to a system providing methodology for moving a project in a complex event processing cluster. 
         [0003]    2. Background Art 
         [0004]    Historically, most businesses were most concerned about the need for accurate storage, retrieval, and analysis of data. Today, businesses are operating in environments where the need to monitor events comes at an ever-accelerating pace. 
         [0005]    Event-driven systems have typically been built around either relational databases or real-time messaging systems, or a combination of both. While these technologies have their advantages, neither is particularly well suited for managing and analyzing events in rapidly changing environments. Relational database servers can process large amounts of stored data and can analyze the information with relative ease but are not designed to operate in real-time environments, and do not provide an effective way to monitor rapidly changing data. Messaging systems permit data to be monitored in real time but generally are not capable of complex computations, correlations, pattern matching or references to historical data. 
         [0006]    Custom applications must often be combined with these technologies to create a viable solution. However, the use of custom applications to compensate for the limitations of these technologies creates new complications. For example, custom applications become increasingly complex very quickly as an organization&#39;s need for progressively more sophisticated analysis grows. Further, custom applications are also costly to modify, and they do not scale well as organizational needs change. 
         [0007]    Tools, such as the Sybase CEP Engine available from Sybase, Inc., of Dublin, Calif., have been developed in the field of Complex Event Processing (CEP) and Event Stream Processing (ESP) applications for use in building applications to process and analyze large volumes of fast-moving data in highly dynamic environments. These tools couple the best features of relational databases and messaging technologies, combined with many new features required for the rapidly changing world faced by today&#39;s businesses. For example, like a real-time messaging system, reactions to each new piece of data occur immediately. Like a relational database system, support exists for the use of a query language similar to SQL to process and analyze incoming data, and to correlate it with historical data. Also, sophisticated pattern-matching capabilities are provided without requiring large amounts of custom application code. 
         [0008]      FIG. 1  illustrates a block diagram of components of an end-to-end application incorporating a complex event processing engine  100 . Messages travel through the application starting from an external data source  110  to an input adapter  120 . The input adapter  120  may poll the external source  110  or register for notifications or use some other mechanism to receive data. The input adapter  120  publishes messages to one or more data streams  130 . The stream(s)  130  feed projects and query module(s)  140 . The query module(s)  140  publish to one or more streams  150 . The stream(s)  150  feed an output adapter  160 . The output adapter  160  subscribes to the stream(s)  150 , processes the messages, converts the data to a format suitable for the destination  170 , and then transmits the output data to the destination  170 . The destination  170  performs whatever actions with the data that it is designed to do. 
         [0009]    Input and output adapters enable the CEP engine  100  to send and receive messages from dynamic and static external sources and destinations. External sources or destinations can include data feeds, sensor devices, messaging systems, radio frequency identification (RFID) readers, e-mail servers, relational databases, and the like. By way of example, a series of input adapters  120  can translate messages from data sources  110 , such as a temperature sensor, bar code scanner and a Java Message Service (JMS) “cloud”, into formats compatible with the CEP engine  100  for processing by several CEP queries  140  with the output adapters  160  converting the CEP result rows into updates to destinations  170 , such as a database server, e-mail messages, and data for a web services dashboard. 
         [0010]    The CEP engine  100  may process hundreds of thousands of messages per second, with latency measured in milliseconds and may run on a single processor, or may consist of multiple servers (i.e., a server cluster) distributed across multiple machines, each containing one or more processors. When a cluster is utilized, a user (or system) may need to control movement of a running CEP project from one node to another in a cluster, such as to balance the load. One basic way to move a project from one node to another would be to stop the running project in the source node and start the project in the destination node. Unfortunately, in this approach, messages which are currently in transition in the CEP may be lost. One way to avoid such losses is to enable persistence in a CEP project, so that the transit stream messages as well as status of the query engine are persisted, such as to persistent storage  180 . However, this will not provide a complete solution, since messages could be in transition in the input  120  and output  160  adapters, and stopping a project may result in those messages being lost. 
         [0011]    By enabling guarantee delivery, which is a communications protocol that provides acknowledgements from the recipient to the sender assuring that messages were delivered, message loss could be avoided, so that, together with persistence enabled, a project could be stopped and moved a project successfully. However, once you add persistence and guarantee delivery, system performance decreases significantly. Performance sensitive CEP projects (e.g., those that avoid the use of disk input/output) may not want to add persistence and guarantee delivery, and some projects may enable persistence and not guarantee delivery. 
         [0012]    Accordingly, a need exists for an approach to provide efficient and successful project movement in a CEP cluster that avoids the problems of the prior art. The present invention addresses such as need. 
       BRIEF SUMMARY 
       [0013]    Briefly stated, the invention includes system, method, computer program product embodiments and combinations and sub-combinations thereof for moving a project in a complex event processing cluster are provided. In an embodiment, project movement in a complex event processing cluster initiates once project persistence capability exists. A handoff of the project from a source node to a destination node in the complex event processing cluster is coordinated by a manager without losing messages and without relying on a message guarantee delivery protocol. 
         [0014]    Further embodiments, features, and advantages of the invention, as well as the structure and operation of the various embodiments of the invention, are described in detail below with reference to accompanying drawings. 
     
    
     
       BRIEF DESCRIPTION OF THE DRAWINGS/FIGURES 
         [0015]    The accompanying drawings, which are incorporated herein and form part of the specification, illustrate embodiments of the present invention and, together with the description, further serve to explain the principles of the invention and to enable a person skilled in the relevant art(s) to make and use the invention. 
           [0016]      FIG. 1  illustrates a block diagram of components of an end-to-end application incorporating a complex event processing engine. 
           [0017]      FIGS. 2   a ,  2   b , and  2   c  illustrate block diagrams of an example of a complex event processing cluster in which a project is moved from one node to another in accordance with an embodiment of the present invention. 
           [0018]      FIG. 3  illustrates a flowchart of an overall process by which a project in a project is moved in a complex event processing cluster in accordance with an embodiment of the present invention. 
           [0019]      FIG. 4  illustrates an example computer useful for implementing components of embodiments of the invention. 
       
    
    
       [0020]    The features and advantages of the present invention will become more apparent from the detailed description set forth below when taken in conjunction with the drawings. In the drawings, like reference numbers generally indicate identical, functionally similar, and/or structurally similar elements. Generally, the drawing in which an element first appears is indicated by the leftmost digit(s) in the corresponding reference number. 
       DETAILED DESCRIPTION 
       [0021]    The present invention relates to a system, method, computer program product embodiments and combinations and sub-combinations thereof for moving a project in a complex event processing cluster. 
         [0022]    In the following description, reference is made concurrently to an example cluster arrangement of CEP servers represented in  FIGS. 2   a ,  2   b , and  2   c  in order to demonstrate changes in connections as a project is moved from one node to another in accordance with an embodiment of the invention, as represented in the block flow diagram of  FIG. 3 . In general, and for purposes of this discussion, a cluster arrangement includes a cluster manager that monitors containers and manages their workload, including coordinating the work among the containers. The work of containers includes not only the running of queries but also the running of streams and adapters. Every cluster suitably contains one or more containers, which are managed by a single active manager. As previously mentioned, the Sybase CEP Engine available from Sybase, Inc., of Dublin, Calif., is suitable for implementing the CEP servers within the cluster arrangement. 
         [0023]    Referring first to  FIG. 2   a , a cluster manager is running in a first node  200 , while containers are running in second node  210  and a third node  220 . An assumption is made that a project currently running in query processor (QP)  230  of node  210  is to be moved to node  220  for running by QP  235  and that a reliable messaging protocol (e.g., guarantee delivery) for the adapters  240 ,  250  is not enabled. The project moving coordination is done by the cluster manager  200 . 
         [0024]    When cluster manager  200  initiates a process to move a project from node  210  to node  220 , such as for load rebalancing, a check is done to determine whether the project has persistence enabled (block  310 ,  FIG. 3 ), and when not enabled, enables it (block  320 ). This is referred to herein as “just in time” persistence and will let the project run till a state of the project is completely persisted. 
         [0025]    With persistence active, the cluster manager  200  then pauses the current project by pausing the input adapter  240  (block  330 ), represented by  FIG. 2   b , in which the arrow link between the input adapter  240  and node  210  has been removed. Once the incoming messages are paused, the query processor  230  will only need to process the existing messages in the system. Input adapters  240  will buffer the messages and wait for instruction to deliver to the query processor  235 . As further shown in  FIG. 2   b , a JIP (just in time persistence) module  260  is represented for providing the data persistence (storing stream messages, messages in transit and status of the query processing/operation). Connections to a disk memory  270  and to system memory  280  are also represented, since, depending on the configuration, the data persistence can be done in the disk  270  or in the memory  280  (such as distributed cache), as is well understood in the art. Persisting in memory is expected to be faster than disk and may be desirable for better performance, since the faster the persistence operation, the shorter the “pause” time for input adapters. 
         [0026]    Then, the destination node for the project, node  220 , is requested to start the project without adapters (block  340 ). This ensures there are no input messages into the query graph, with the project instantiated, but not running, in the destination node  220 . In this state, the project is synchronized with the persisted data from the source node  210  (block  350 ) once the persistence operation is completed in the source node  210 . The persisted messages are read from either from the disk  270  or memory  280 , depending upon the configuration. Once the synch up is completed, the project at the source node  210  is stopped (block  360 ) and the adapters (output and input) of the project in the destination node  220  are started (block  370 ), as represented by the set of connections illustrated via the arrows in  FIG. 2   c . In this manner, the transfer of a project from one node to another in a CEP system is successfully completed in an efficient and accurate manner without significantly impacting system performance. 
         [0027]    Some consideration may need to be made if the input and output adapters are “out-of-process” adapters, which refers to adapters that run as a separate process, rather than as part of the server. For an out-of-process input adapter to connect to an input stream and write data to that stream, the adapter must know the URI (Uniform Resource Identifier) of that stream. Similarly, an out-of-process output adapter must know the URI of the stream from which it will read. As is generally understood, each data stream has a unique stream URI, which provides enough information that a program can connect to that stream. 
         [0028]    Thus, while the cluster manager  200  can manage the life cycle (start, pause, stop etc.) of in-process adapters, for out-of-process ones, the cluster manager  200  may not able to do it. In such a situation, an output-of-process adapter needs to have awareness of the possibility of moving projects inside the cluster without its knowledge and needs to be able to resolve the URI from the cluster manager  200  whenever the destination project does not appear to be running in the destination node. In an embodiment, resolution of the new URI of a project for an out-of-process adapter occurs through reconnection to the cluster manager  200  whenever a connection is lost. It is also important in such situations that an out-of-process input adapter buffer that messages that it is trying to deliver when the connection is lost to the stream and deliver them when connection is reestablished, as is well appreciated by those skilled in the art. 
         [0029]    Various aspects of the present invention can be implemented by software, firmware, hardware, or a combination thereof.  FIG. 4  illustrates an example computer system  400 , such as capable of acting as the servers in the cluster of  FIGS. 2   a ,  2   b , and  2   c , in which the present invention, or portions thereof, can be implemented as computer-readable code. For example, the methods illustrated by flowchart of  FIG. 3  can be implemented in system  400 . Various embodiments of the invention are described in terms of this example computer system  400 . After reading this description, it will become apparent to a person skilled in the relevant art how to implement the invention using other computer systems and/or computer architectures. 
         [0030]    Computer system  400  includes one or more processors, such as processor  404 . Processor  404  can be a special purpose or a general purpose processor. Processor  404  is connected to a communication infrastructure  406  (for example, a bus or network). 
         [0031]    Computer system  400  also includes a main memory  408 , preferably random access memory (RAM), and may also include a secondary memory  410 . Secondary memory  410  may include, for example, a hard disk drive  412 , a removable storage drive  414 , and/or a memory stick. Removable storage drive  414  may comprise a floppy disk drive, a magnetic tape drive, an optical disk drive, a flash memory, or the like. The removable storage drive  414  reads from and/or writes to a removable storage unit  418  in a well known manner. Removable storage unit  418  may comprise a floppy disk, magnetic tape, optical disk, etc. which is read by and written to by removable storage drive  414 . As will be appreciated by persons skilled in the relevant art(s), removable storage unit  418  includes a computer usable storage medium having stored therein computer software and/or data. 
         [0032]    In alternative implementations, secondary memory  410  may include other similar means for allowing computer programs or other instructions to be loaded into computer system  400 . Such means may include, for example, a removable storage unit  422  and an interface  420 . Examples of such means may include a program cartridge and cartridge interface (such as that found in video game devices), a removable memory chip (such as an EPROM, or PROM) and associated socket, and other removable storage units  422  and interfaces  420  which allow software and data to be transferred from the removable storage unit  422  to computer system  400 . 
         [0033]    Computer system  400  may also include a communications interface  424 . 
         [0034]    Communications interface  424  allows software and data to be transferred between computer system  400  and external devices. Communications interface  424  may include a modem, a network interface (such as an Ethernet card), a communications port, a PCMCIA slot and card, or the like. Software and data transferred via communications interface  424  are in the form of signals which may be electronic, electromagnetic, optical, or other signals capable of being received by communications interface  424 . These signals are provided to communications interface  424  via a communications path  426 . Communications path  426  carries signals and may be implemented using wire or cable, fiber optics, a phone line, a cellular phone link, an RF link or other communications channels. 
         [0035]    In this document, the terms “computer program medium” and “computer usable medium” are used to generally refer to media such as removable storage unit  418 , removable storage unit  422 , and a hard disk installed in hard disk drive  412 . Signals carried over communications path  426  can also embody the logic described herein. Computer program medium and computer usable medium can also refer to memories, such as main memory  408  and secondary memory  410 , which can be memory semiconductors (e.g. DRAMs, etc.). These computer program products are means for providing software to computer system  400 . 
         [0036]    Computer programs (also called computer control logic) are stored in main memory  408  and/or secondary memory  410 . Computer programs may also be received via communications interface  424 . Such computer programs, when executed, enable computer system  400  to implement the present invention as discussed herein. In particular, the computer programs, when executed, enable processor  404  to implement the processes of the present invention, such as the method illustrated by the flowchart of  FIG. 3 . Accordingly, such computer programs represent controllers of the computer system  400 . Where the invention is implemented using software, the software may be stored in a computer program product and loaded into computer system  400  using removable storage drive  414 , interface  420 , hard drive  412  or communications interface  424 . 
         [0037]    The invention is also directed to computer program products comprising software stored on any computer useable medium. Such software, when executed in one or more data processing device, causes a data processing device(s) to operate as described herein. Embodiments of the invention employ any computer useable or readable medium, known now or in the future. Examples of computer useable mediums include, but are not limited to, primary storage devices (e.g., any type of random access memory), secondary storage devices (e.g., hard drives, floppy disks, CD ROMS, ZIP disks, tapes, magnetic storage devices, optical storage devices, MEMS, nanotechnological storage device, etc.), and communication mediums (e.g., wired and wireless communications networks, local area networks, wide area networks, intranets, etc.). 
         [0038]    While various embodiments of the present invention have been described above, it should be understood that they have been presented by way of example only, and not limitation. It will be understood by those skilled in the relevant art(s) that various changes in form and details may be made therein without departing from the spirit and scope of the invention as defined in the appended claims. The invention is applicable to any elements operating as described herein. Accordingly, the breadth and scope of the present invention should not be limited by any of the above-described exemplary embodiments, but should be defined only in accordance with the following claims and their equivalents.