[FLINK-4520] Integrate Siddhi as a lightweight CEP Library

• Type: New Feature
• Status: Resolved
• Priority: Major
• Resolution: Resolved
• Affects Version/s: 1.2.0
• Component/s: Library / CEP
• Labels: cep library patch-available

Release Note

Merged Bahir Flink:
* https://issues.apache.org/jira/browse/BAHIR-144
* https://github.com/apache/bahir-flink/pull/22

• Assignee: Hao Chen
• Reporter: Hao Chen
• Created: 28/Aug/16 02:59
• Updated: 27/Nov/17 02:44
• Resolved: 27/Nov/17 02:42

Description

flink-siddhi proposal

Abstraction

Siddhi CEP is a lightweight and easy-to-use Open Source Complex Event Processing Engine (CEP) released as a Java Library under `Apache Software License v2.0`. Siddhi CEP processes events which are generated by various event sources, analyses them and notifies appropriate complex events according to the user specified queries.

It would be very helpful for flink users (especially streaming application developer) to provide a library to run Siddhi CEP query directly in Flink streaming application.

• http://wso2.com/products/complex-event-processor/
• https://github.com/wso2/siddhi

Features

• Integrate Siddhi CEP as an stream operator (i.e. `TupleStreamSiddhiOperator`), supporting rich CEP features like

• Filter
• Join
• Aggregation
• Group by
• Having
• Window
• Conditions and Expressions
• Pattern processing
• Sequence processing
• Event Tables
  ...

• Provide easy-to-use Siddhi CEP API to integrate Flink DataStream API (See `SiddhiCEP` and `SiddhiStream`)
• Register Flink DataStream associating native type information with Siddhi Stream Schema, supporting POJO,Tuple, Primitive Type, etc.
• Connect with single or multiple Flink DataStreams with Siddhi CEP Execution Plan
• Return output stream as DataStream with type intelligently inferred from Siddhi Stream Schema

• Integrate siddhi runtime state management with Flink state (See `AbstractSiddhiOperator`)

• Support siddhi plugin management to extend CEP functions. (See `SiddhiCEP#registerExtension`)

Test Cases

• org.apache.flink.contrib.siddhi.SiddhiCEPITCase: https://github.com/haoch/flink/blob/FLINK-4520/flink-contrib/flink-siddhi/src/test/java/org/apache/flink/contrib/siddhi/SiddhiCEPITCase.java

Example

 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
 SiddhiCEP cep = SiddhiCEP.getSiddhiEnvironment(env);

 cep.registerExtension("custom:plus",CustomPlusFunctionExtension.class);

 cep.registerStream("inputStream1", input1, "id", "name", "price","timestamp");
 cep.registerStream("inputStream2", input2, "id", "name", "price","timestamp");

 DataStream<Tuple5<Integer,String,Integer,String,Double>> output = cep
  .from("inputStream1").union("inputStream2")
  .sql(
    "from every s1 = inputStream1[id == 2] "
     + " -> s2 = inputStream2[id == 3] "
     + "select s1.id as id_1, s1.name as name_1, s2.id as id_2, s2.name as name_2 , custom:plus(s1.price,s2.price) as price"
     + "insert into outputStream"
  )
  .returns("outputStream");

 env.execute();

Issue Links

• links to: BAHIR-144
• links to: GitHub Pull Request #2487
• links to: GitHub Pull Request bahir-flink#22