A light-weight library to run Siddhi CEP within Apache Flink streaming application.
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.
This project is mainly to provide a light-weight library to easily run Siddhi CEP within flink streaming application.
TupleStreamSiddhiOperator
), supporting rich CEP features like
SiddhiCEP
and SiddhiStream
)
AbstractSiddhiOperator
)SiddhiCEP#registerExtension
)1.8
)1.7.0
)git clone git@github.com:haoch/flink-siddhi.git
mvn clean install -DskipTests
mvn clean test
Add com.github.haoch:flink-siddhi
in project dependencies:
<dependencies>
<dependency>
<groupId>com.github.haoch</groupId>
<artifactId>flink-siddhi_2.11</artifactId>
<version>LATEST</version>
</dependency>
</dependencies>
<repositories>
<repository>
<id>clojars</id>
<url>http://clojars.org/repo/</url>
</repository>
</repositories>
Execute SiddhiQL
with SiddhiCEP
API, for 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")
.cql(
"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();
For more examples, please see
org.apache.flink.contrib.siddhi.SiddhiCEPITCase
Welcome to make contribution to code or document by sending a pull request, or reporting issues or bugs.
Licensed under the Apache License, Version 2.0. More details, please refer to LICENSE file.