Parallel processing of continuous queries on data streams

Abstract

A continuous query parallel engine on data streams provides scalability and increases the throughput by the addition of new nodes. The parallel processing can be applied to data stream processing and complex events processing. The continuous query parallel engine receives the query to be deployed and splits the original query into subqueries, obtaining at least one subquery; each subquery is executed in at least in one node. Tuples produced by each operator of each subquery are labeled with timestamps. A load balancer is interposed at the output of each node that executes each one of the instances of the source subquery and an input merger is interposed in each node that executes each one of the instances of a destination subquery. After checks are performed, further load balancers or input managers may be added.

1 . A parallel stream processing engine of continuous queries formed by a plurality of instances wherein each instance is executed in any processing node that of the processing engine, wherein the cooperation of instances processes a query, comprising:
 a) means for receiving the query to be deployed;   b) means for splitting the original query into subqueries, obtaining at least one subquery; each subquery is executed at least in one node;   c) means for labeling with timestamps the tuples produced by each operator of each subquery;   d) wherein between each two consecutive subqueries,
 i) a load balancer is interposed at the output of each node, the load balancer executing each of the instances of the source subquery; 
 ii) the output from the load balancer is connected with all the nodes in which one of the instances of the destination subquery is executed; 
 iii) an input merger is interposed in each node, the input merger executing each of the instances of destination subquery; 
 iv) checking is performed to determine if all subqueries contain at most a stateful operator and if inputs are connected to previous subqueries, wherein:
 a) if the checking succeeds, the load balancer sends received tuples with a same key to a same instance of the source subquery; 
 b) if the checking fails, a load balancer and an input merger are interposed before each stateful operator that is preceded by any other operator, such that the stateful operator in the node where each of the instances of the destination query are executed sends all received tuples with the same key to the same instance of the destination subquery. 
 
   
     
     
         2 . The parallel stream processing engine of continuous queries, according to  claim 1 , wherein the means for splitting queries uses one of the following methods:
 a) non splitting the query, being the result a single subquery identical to the original whole query;   b) splitting the source query into as many subqueries as operators included in the original query;   c) splitting the source query into subqueries, where each subquery consists of one stateful operator followed by at least a stateless operator, except an initial subquery only contains stateless operators;   d) splitting the Source query into subqueries following another criterion.   
     
     
         3 . The parallel stream processing engine of continuous queries, according to  claim 1 , wherein the sending of tuples received with the same key to the same instance of the source subquery comprises:
 a) applying to each tuple with key c a hash function obtaining a value h=hash;   b) obtaining the partition identifier p of the tuple, making the modulo operation by a constant np, obtaining a value p=h mod np;   c) assigning each partition with identifier p to an instance of the subqueries.   
     
     
         4 . The parallel stream processing engine of continuous queries, according to  claim 1 , wherein the input merger forwards tuples as the input merger receives the tuples from any of the source subqueries to the instance of the destination subquery is connected to. 
     
     
         5 . The parallel stream processing engine of continuous queries, according to  claim 1 , wherein the input merger waiting to receive a tuple from each one of the input merger input streams before forwarding the tuple with smallest timestamp to the destination subquery is connected to. 
     
     
         6 . The parallel stream processing engine of continuous queries, according to the  claim 5 , wherein a load balancer storing the timestamp from the last generated tuple for each destination subquery and if after a maximum prefixed period of time m, a tuple has not been sent to the destination subquery, the load balancer sends a dummy tuple with the same timestamp the last tuple sent by the load balancer had. 
     
     
         7 . The parallel processing engine of continuous queries on data streams, according to  claim 3 , wherein the engine is able to reconfigure the processing of tuples corresponding to a partition p, by transferring the processing of those tuples from a source instance to a destination instance, comprising:
 a) obtaining the greatest active timestamp in the system, mt;   b) establishing a future timestamp mtf, by adding temporal shift fεN to the greatest active timestamp in the system, mt, mtf=mt+f;   c) sending, during the established time margin f, a reconfiguration command including the timestamp that begins the reconfiguration, mtf, to all source instances involved in the reconfiguration;   d) storing the timestamp mtf in all load balancers of the source subquery;   e) sending of the tuples corresponding to the reconfigured partition p with timestamp less or equal than mtf by the load balancers of the source subquery to the instance of the destination subquery responsible for the partition p before the reconfiguration started;   f) sending of one tuple by the load balancers of the source subquery that indicates the end of reconfiguration to the instances of the destination subquery involved in the state transfer before sending the first tuple from partition p greater than mtf;   g) sending of the tuples belonging to reconfigured partition p with timestamp greater than mtf by the load balancers of the source subquery to the instance of the destination subquery responsible for the partition p after the reconfiguration;   h) transferring the state of partition p to the destination instance after receiving the tuple that indicates the end of reconfiguration from all load balancers of the source subquery;   i) apply at the destination instance the state of the partition p after receiving the state transfer from partition p   
     
     
         8 . The parallel stream processing engine of continuous queries, according to  claim 7 , wherein the configuration of one of the nodes where a subquery is executed to balance the load comprises:
 a) receiving periodically from all the nodes where the query is being executed, data about CPU and memory utilization at each node, and the greatest timestamp among the processed tuples at each node.   b) comparing utilization data across the nodes executing the query;   c) if, for a subquery, the comparison between utilization exceeds a first given upper utilization threshold, the node reconfigures at least a data partition, sending a reconfiguration command to the involved instances of the source and destination subquery;   d) if the average load of the nodes of a subquery exceeds a given second upper imbalance threshold, the node selects one node from the set of free nodes, adds the selected node to the set of nodes that executes the subquery so that, the selected node will receive load from other nodes automatically;   e) if the average load of the nodes of a subquery can be satisfied with a smaller set of nodes without exceeding the second upper utilization threshold, the node selects one node from the set of nodes where the subquery is being executed, and sends a reconfiguration command to the selected node so all the selected nodes partitions will be distributed among the rest of the nodes that execute the subquery.   
     
     
         9 . A method of parallel stream processing continuous queries formed by a plurality of instances wherein each instance is executed in a processing node comprising a processing engine, wherein the cooperation of instances processes a query, comprising the following steps:
 a) receiving the query to be deployed;   b) splitting the original query into subqueries, obtaining at least one subquery; each subquery is executed at least in one node;   c) labeling with timestamps the tuples produced by each operator of each subquery;   d) between each two consecutive subqueries,
 i) interposing a load balancer at the output of each node, the load balancer executing each of the instances of the source subquery; 
 ii) connetcting the output from the load balancer with all the nodes in which one of the instances of the destination subquery is executed; 
 iii) interposing an input merger in each node, the input merger executing each of the instances of destination subquery; 
 iv) checking if all subqueries contain at most a stateful operator and if inputs are connected to previous subqueries, wherein:
 a) if the checking succeeds, the load balancer sends received tuples with a same key to a same instance of the source subquery; 
 b) if the checking fails, a load balancer and an input merger are interposed before each stateful operator that is preceded by any other operator, such that the stateful operator in the node where each of the instances of the destination query are executed sends all received tuples with the same key to the same instance of the destination subquery. 
 
   
     
     
         10 . The method of parallel stream processing continuous queries, according to  claim 9 , wherein the means for splitting queries is made by one of the following methods:
 a) non splitting the query, being the result a single subquery identical to the original whole query;   b) splitting the source query into as many subqueries as operators included in the original query;   c) splitting the source query into subqueries, where each subquery consists of one stateful operator followed by at least a stateless operator, except an initial subquery only contains stateless operators;   d) splitting the Source query into subqueries following another criterion.   
     
     
         11 . The method of parallel stream processing continuous queries, according to  claim 9 , wherein the sending of tuples received with the same key to the same instance of the source subquery comprises:
 a) applying to each tuple with key c a hash function obtaining a value h=hash;   b) obtaining the partition identifier p of the tuple, making the modulo operation by a constant np, obtaining a value p=h mod np;   c) assigning each partition with identifier p to an instance of the subqueries.   
     
     
         12 . The method of parallel stream processing continuous queries, according to  claim 9 , wherein the input merger forwards tuples as the input merger receives the tuples from any of the source subqueries to the instance of the destination subquery is connected to. 
     
     
         13 . The method of parallel stream processing continuous queries, according to  claim 9 , wherein the input merger waiting to receive a tuple from each one of the input merger input streams before forwarding the tuple with smallest timestamp to the destination subquery is connected to. 
     
     
         14 . The method of parallel stream processing continuous queries, according to  claim 13 , wherein a load balancer storing the timestamp from the last generated tuple for each destination subquery and if after a maximum prefixed period of time m, a tuple has not been sent to the destination subquery, the load balancer sends a dummy tuple with the same timestamp the last tuple sent by the load balancer had. 
     
     
         15 . The method of parallel processing continuous queries on data streams, according to  claim 11 , wherein the engine is able to reconfigure the processing of tuples corresponding to a partition p, by transferring the processing of those tuples from a source instance to a destination instance, comprising:
 a) obtaining the greatest active timestamp in the system, mt;   b) establishing a future timestamp mtf, by adding temporal shift fεN to the greatest active timestamp in the system, mt, mtf=mt+f;   c) sending, during the established time margin f, a reconfiguration command including the timestamp that begins the reconfiguration, mtf, to all source instances involved in the reconfiguration;   d) storing the timestamp mtf in all load balancers of the source subquery;   e) sending of the tuples corresponding to the reconfigured partition p with timestamp less or equal than mtf by the load balancers of the source subquery to the instance of the destination subquery responsible for the partition p before the reconfiguration started;   f) sending of one tuple by the load balancers of the source subquery that indicates the end of reconfiguration to the instances of the destination subquery involved in the state transfer before sending the first tuple from partition p greater than mtf;   g) sending of the tuples belonging to reconfigured partition p with timestamp greater than mtf by the load balancers of the source subquery to the instance of the destination subquery responsible for the partition p after the reconfiguration;   h) transferring the state of partition p to the destination instance after receiving the tuple that indicates the end of reconfiguration from all load balancers of the source subquery;   i) apply at the destination instance the state of the partition p after receiving the state transfer from partition p   
     
     
         16 . The method of parallel stream processing continuous queries, according to  claim 15 , wherein the configuration of one of the nodes where a subquery is executed to balance the load comprises:
 a) receiving periodically from all the nodes where the query is being executed, data about CPU and memory utilization at each node, and the greatest timestamp among the processed tuples at each node.   b) comparing utilization data across the nodes executing the query;   c) if, for a subquery, the comparison between utilization exceeds a first given upper utilization threshold, the node reconfigures at least a data partition, sending a reconfiguration command to the involved instances of the source and destination subquery;   d) if the average load of the nodes of a subquery exceeds a given second upper imbalance threshold, the node selects one node from the set of free nodes, adds the selected node to the set of nodes that executes the subquery so that, the selected node will receive load from other nodes automatically;   e) if the average load of the nodes of a subquery can be satisfied with a smaller set of nodes without exceeding the second upper utilization threshold, the node selects one node from the set of nodes where the subquery is being executed, and sends a reconfiguration command to the selected node so all the selected nodes partitions will be distributed among the rest of the nodes that execute the subquery.