WSO2 Complex Event Processor 4.1

WSO2 Complex Event Processor  4.1 released  23rd February 2016.

This release mainly focuses on improving and stabilizing the product and enhancing its capabilities.

One of the main features that was included in this release is instrumenting and monitoring support for WSO2CEP  as well as Siddhi.  This enable users to identify throughput and memory consumption of each and every component of WSO2CEP and Siddhi. Through this users can identify possible bottlenecks in their queries and optimize CEP for better performance.

The same is also exposed via JMX  such that It can be monitored to third party JMX consumers  such as jconsole.

Another important feature CEP introduced in this release is visualizing Siddhi queries.

One of the notable improvement of this release is its improved high-availability support. now CEP can support high availability with more than two nodes providing zero downtime with no data loss, and as it also stores its state as a periodic snapshot to a database even during full cluster restart CEP has the capability of restoring it state from its last available snapshot. For more information refer the documentation here.

Further WSO2CEP has introduced several improvements to it’s core runtime complex event processor engine Siddhi. They are as follows:

• Hazelcast Event Table - Allowing events to be stored and  manipulated in  Hazelcast based In-Memory Data Grid.
• Minima and Maxima detection - Detecting maxima and minima in an event pattern, this allows detecting complex  stock market patterns using combinations of maxima and minima.
• Map extension -  This is introduced to support arbitrary key value pairs in Siddhi which was not supported for a long time. This function allows users to create a map, add remove and check for keys and values within a hash map.
• InsertOrUpdate to Event Table -  This facilitate doing and insert or an update as an atomic operation on Even Tables.
• Outer and left joins in Siddhi - In addition to inner joins Siddhi now supports outer and left joins
• Time length window - The eviction policy on Siddhi window is triggered both by time and length properties.
• External Batch window -  Allowing batch window to get triggered by event time rather by getting triggered by the system time.

You can download WSO2 Complex Event processor 4.1 from here, its documentation from here,  and find the latest Siddhi documentation here.

Adding Statistics to Siddhi 3

A new statistics feature is added to Siddhi 3.0.5.

This will help users to find how their queries are behaving and to understand what query to optimise to get better performance and to optimise memory consumption.

This is very easy to enable and you can enable this by just adding

"@plan:statistics(reporter = 'console', interval = '5' )"

to the execution plan will start statistics reporting on console every 5 seconds.

You will be able to and output something like

1/12/16 12:01:39 AM ============================================================

-- Gauges ----------------------------------------------------------------------
org.wso2.siddhi.executionplan.5910c3f3-1af3-43bd-ae6e-06b1a959eaf3.query1.memory
             value = 548804
org.wso2.siddhi.executionplan.5910c3f3-1af3-43bd-ae6e-06b1a959eaf3.query2.memory
             value = 556686

-- Meters ----------------------------------------------------------------------
org.wso2.siddhi.stream.cseEventStream.throughput
             count = 2
         mean rate = 0.37 events/second
     1-minute rate = 0.40 events/second
     5-minute rate = 0.40 events/second
    15-minute rate = 0.40 events/second
org.wso2.siddhi.stream.cseEventStream2.throughput
             count = 0
         mean rate = 0.00 events/second
     1-minute rate = 0.00 events/second
     5-minute rate = 0.00 events/second
    15-minute rate = 0.00 events/second
org.wso2.siddhi.stream.outputStream.throughput
             count = 3
         mean rate = 0.57 events/second
     1-minute rate = 0.60 events/second
     5-minute rate = 0.60 events/second
    15-minute rate = 0.60 events/second

-- Timers ----------------------------------------------------------------------
org.wso2.siddhi.executionplan.5910c3f3-1af3-43bd-ae6e-06b1a959eaf3.query1.latency
             count = 2
         mean rate = 0.38 calls/second
     1-minute rate = 0.40 calls/second
     5-minute rate = 0.40 calls/second
    15-minute rate = 0.40 calls/second
               min = 0.01 milliseconds
               max = 0.08 milliseconds
              mean = 0.05 milliseconds
            stddev = 0.04 milliseconds
            median = 0.08 milliseconds
              75% <= 0.08 milliseconds
              95% <= 0.08 milliseconds
              98% <= 0.08 milliseconds
              99% <= 0.08 milliseconds
            99.9% <= 0.08 milliseconds
org.wso2.siddhi.executionplan.5910c3f3-1af3-43bd-ae6e-06b1a959eaf3.query2.latency
             count = 2
         mean rate = 0.38 calls/second
     1-minute rate = 0.40 calls/second
     5-minute rate = 0.40 calls/second
    15-minute rate = 0.40 calls/second
               min = 0.01 milliseconds
               max = 0.01 milliseconds
              mean = 0.01 milliseconds
            stddev = 0.00 milliseconds
            median = 0.01 milliseconds
              75% <= 0.01 milliseconds
              95% <= 0.01 milliseconds
              98% <= 0.01 milliseconds
              99% <= 0.01 milliseconds
            99.9% <= 0.01 milliseconds

You can also add

"@plan:statistics(reporter = 'jmx')"

to report stats via jmx.

You can find a sample implementations here.
https://github.com/wso2/siddhi/blob/master/modules/siddhi-core/src/test/java/org/wso2/siddhi/core/managment/StatisticsTestCase.java

Siddhi also allows users to add custom reporting tools. To do so.
Just implement the StatisticsTrackerFactory interface

package org.wso2.siddhi.core.util.statistics;
import org.wso2.siddhi.query.api.annotation.Element;
import java.util.List;

public interface StatisticsTrackerFactory {
    LatencyTracker createLatencyTracker(String name, StatisticsManager statisticsManager);
    ThroughputTracker createThroughputTracker(String name, StatisticsManager statisticsManager);
    MemoryUsageTracker createMemoryUsageTracker(StatisticsManager statisticsManager);
    StatisticsManager createStatisticsManager(List<Element> elements);
}

And add that to the Siddhi Manager

siddhiManager.setStatisticsConfiguration(new StatisticsConfiguration(new MyMetricsFactory()));

Happy Coding :) 

What to expect in WSO2 CEP 4.0

WSO2 Complex Event Processor 4.0 is released! A long waited CEP release from WSO2. This is a major break through among current CEP products.

Here are some notable features it has introduced :

Scale and Performance 

1. Support for distributed processing with Apache Storm - WSO2 CEP runs Siddhi CEP on Apache Storm in a distributed mode while providing an SQL like Siddhi Query Language to define queries on Storm. 
2. Faster CEP engine with WSO2 Siddhi 3.0 - With LMAX Disruptor and improved memory management, Siddhi was able to run even faster than its predecessors doing ~ 4M Events/Sec
3. HA Support - Supporting Hot-Worm High Availability with zero downtime
4. More effective native thrift/binary transports 

Usability 

1. Realtime Dashboard support - Support for gadget generation wizard and customisable realtime dashboard
2. Easy to use management console for business users - Ability to tweak queries with a form based UI without knowing the implementation and complexities of SiddhiQL, Storm and  Java 
3. Event simulation capability - Try sending events easily 
4. Siddhi query try-it - Try writing queries 
5. Event flow visualisation - See how you the event flow pipeline will look like

Enhancements  

1. New protocols introduced such as MQTT, Kafka, Websocket - Support for many new and improved event receiver and publisher capabilities
2. Enhanced query language with more extensions -  Such as Function, Stream Processor, Window and Aggregate Function Processor
3. Support for machine learning via ml, pmml (Predictive Model Markup Language) extensions - To run machine learning model on realtime
4. Support for Natural Language processing via nlp (Natural Language Processing) extension - Supported via Stanford NLP
5. Geo processing capabilities via geo extension and Geo Dashboard - Dashboard having the capability to monitor and apply queries on geo spaces. 
6. Support for scripting using JavaScript, Scala and R - Can use scripts to write CEP logic!

This release not only overcomes the scaling bottleneck of previous WSO2 CEP versions, but it also provides a seamless query language that allows uses to run the same queries both on single node and on distributed environment, making WSO2 CEP 4.0 as a unique product in the market. 

You can download WSO2 CEP 4.0 from here 

and a quick getting started guide can be found here 

Try it out, and for questions please post on StackOverflow with tag 'wso2cep'. The community and the WSO2 team will be very happy to answer :)

You can contribute your code e.g. extensions and adopters, such that it will be maintained and released on future product releases:

For Extensions and Siddhi improvements please send the pull request to Siddhi repo and 

For adopters and other contributions please send to carbon-analytics-common carbon-event-processing and product-cep repos :) 

Be Realtime !!

Adding Siddhi extensions when using Siddhi as a Java library

You can write Siddhi extensions and run them in WSO2 CEP as given : https://docs.wso2.com/display/CEP310/Writing+Extensions+to+Siddhi

But if you are using Siddhi as a Java library then you can add then to SiddhiManager as follows

        List extensionClasses = new ArrayList();

        extensionClasses.add(org.wso2.siddhi.core.query.processor.window.UniqueWindowProcessorExt.class);

        

        SiddhiConfiguration siddhiConfiguration = new SiddhiConfiguration();

        siddhiConfiguration.setSiddhiExtensions(extensionClasses);

        

        SiddhiManager siddhiManager = new SiddhiManager(siddhiConfiguration);  

Getting inside out of Social Data using WSO2 BAM & WSO2 CEP

We engage in Social Media, but have we ever analysed the data it contains?

Why we need to analyse?
Can it help us to make prompt business decisions?
How we can do that?

I have done a screencast to answer all these,
using WSO2 Business Activity Monitor (2.4.0) and WSO2 Complex Event Processor (3.0.0).

Using Siddhi CEP as a Java library

WSO2 Complex Event processor uses Siddhi as the Backend Runtime Engine, which was initially developed at http://siddhi.sourceforge.net/, and its development is now being continued as a commons project of WSO2 Inc.

Siddhi under Apache License V2, and its a product designed for real time processing both in standalone mode and in distributed mode.

Siddhi is basically a Java library that's embedded in WSO2 CEP server to provide enterprise integration for Complex Event Processing.

Since Siddhi is just a java library here are some samples demonstrating siddhi’s capabilities as a library.

You can find the WSO2 CEP product from: 

http://wso2.com/products/complex-event-processor/

You can find the Siddhi source code from: 

https://svn.wso2.org/repos/wso2/carbon/platform/branches/4.0.0/dependencies/commons/siddhi/1.0.1-wso2v1

You can find the Documentation for WSO2 CEP from: 

http://docs.wso2.org/wiki/display/CEP201/Complex+Event+Processor+Documentation

You can find the Documentation for Siddhi Language Specification from: 

http://docs.wso2.org/wiki/display/CEP201/Siddhi+Language+Specification

You can find the Siddhi Library Sample from: 

https://svn.wso2.org/repos/wso2/people/suho/packs/siddhi/siddhi-1.0.1-wso2v1-samples.zip

Distributed Processing Sample for WSO2 CEP

Today we released WSO2 Complex Event Processor 2.0.0 Milestone 2.
This is available at  https://svn.wso2.org/repos/wso2/people/suho/packs/cep/wso2cep-2.0.0-M2.zip

One of the key feature of this CEP is its support for distributed processing via Siddhi CEP Engine. To demonstrate its capability I came up with a sample on distributed Processing. This sample uses Siddhi CEP Engine for processing and JMS ActiveMQ Broker to publish and subscribe events.

To run the Distributed Processing Sample follow the steps give below;
 
Step 01: Configure and run ActiveMQ in your local machine

Download the ActiveMQ from "http://activemq.apache.org/activemq-543-release.html".
unzip the distribution and run the ActiveMQ server using the command “./activemq console” from apache-activemq-xxx/bin (in Linux)
Note: WSO2CEP has been tested with ActiveMQ 5.4.3

For each CEP node in the cluster follow the steps from 02 to 08.

Step 02: Deploy CEP server 

Unzip the CEP server (Do not start the server).

      
Step 03: Change the CEP server Offset.

If you are running multiple servers in the same machine change the offset
from file "wso2cep-2.0.0-2/repository/conf/carbon.xml" to different numbers. 

E.g.

<offset>1</offset>

This is to overcome server port conflicts.
E.g. If three WSO2 servers are going to be deployed in the same machine, they can have offsets as 1, 2, and 3.

Step 04: Copy paste ActiveMQ jars.

Copy paste activemq-all-xxx.jar from the ActiveMQ home directory to 

wso2cep-2.0.0/samples/lib directory.

Copy paste activemq-core-xxx.jar and geronimo-j2ee-management_1.1_spec-1.0.1.jar
from apache-activemq-xxx/lib to wso2cep-2.0.0/repository/components/lib directory

 
Step 06: Deploy required broker management configuration

This allows CEP to receive and send messages via JMS Broker.

To deploy run "ant deploy-broker-manager" from wso2cep-2.0.0/samples/cep-samples directory.

Step 07: Configure input, output and queries

For this sample, the configurations of input, output and queries are at wso2cep-2.0.0-1/samples/cep-samples/conf/buckets/purchase-analyser-bucket.xml
When running multiple server nodes each server need to receive the input events through different topics. To enable this behaviour, open the above file and change the input topics to have different names. The default input topic name is “PurchaseTopic”.

E.g changing the names to “PurchaseTopic1”.

<input brokername="activemqJmsBroker" topic="PurchaseTopic1" />

Note : To enable distributed processing (already enabled for this sample) the change we have to do is to make the "siddhi.enable.distributed.processing" property to "true".

<engineproviderconfiguration engineprovider="SiddhiCEPRuntime">
     <property name="siddhi.persistence.snapshot.time.interval.minutes">0</property>
     <property name="siddhi.enable.distributed.processing">true</property>
</engineproviderconfiguration>

Step 08: Deploy bucket configuration

Deploy the bucket by adding the configuration purchase-analyser-bucket.xml to the directory wso2cep-2.0.0/repository/deployment/server/cepbuckets/

Step 09: Start all the servers

Start CEP servers by running "./wso2server.sh" from wso2cep-2.0.0/bin directory 

Step 10:Subscribing to output topic

Start the subscriber of the output topic "PurchaseResults" by running "ant purchaseResultsSubscriber" in a separate terminal,
from wso2cep-2.0.0/samples/cep-samples directory

Step 11:Publishing events

To publish events to all the servers (to their input topics as we defined in Step 07 ) run the publisher from wso2cep-2.0.0/samples/cep-samples directory
with the command  "ant purchasePublisher -Dtopics=xxx,xxx,..."
The client will publish events to all the given topics in a round robin manner.

E.g If we we have configured the buckets to receive the inputs via the input topics as PurchaseTopic1, PurchaseTopic2 and PurchaseTopic3

"ant purchasePublisher -Dtopics=PurchaseTopic1,PurchaseTopic2,PurchaseTopic3."

Step 12: Observation

You will be able to observe how the counts in the results steadily increases when messages are sent to any of the servers, and how the results remain consistent even when some servers goes down and comes back.