flink-siddhi
flink-siddhi copied to clipboard
haoch
A CEP library to run Siddhi within Apache Flink™ Streaming Application
Hi @haoch How the operator state behaves in parallelism? As I can see in the class `org.apache.flink.streaming.siddhi.operator.AbstractSiddhiOperator` we are using the managedOperatorState. Is there a way to partition the rules...
**POM**: ``` 4.0.0 flinksidhi flinksidhipoc 1.0-SNAPSHOT flinksidhipoc http://www.example.com UTF-8 1.7 1.7 5.1.2 1.9.0 junit junit 4.11 test org.apache.flink flink-streaming-java_2.11 ${flink.version} org.apache.flink flink-connector-kafka_2.11 ${flink.version} org.apache.flink flink-clients_2.11 ${flink.version} com.github.haoch flink-siddhi_2.11 LATEST io.siddhi...
I find Siddhi very interesting but i am concerned with the lack of recent activity, is there less activity because both project are stable ? Is this project future proof...
With AWS Kinesis Flink 1.8 Flink application built on flink 1.8.2 and flink-siddhi 0.2.2-SNAPSHOT or 0.2.1 both versions failed with below error: messageType INFO threadName flink-akka.actor.default-dispatcher-2 throwableInformation.0 **java.lang.AbstractMethodError** throwableInformation.1 at...
PR corrects the situation when data stream events are assigned to partitions by AddRouteOperator, but control events aren't. This happens because control events can't be distributed on broadcast rule through...
https://github.com/haoch/flink-siddhi/blob/cf4b2b6935bc69be1b465e49431bb8485e4d670b/core/src/main/java/org/apache/flink/streaming/siddhi/operator/AbstractSiddhiOperator.java#L341 https://github.com/haoch/flink-siddhi/blob/cf4b2b6935bc69be1b465e49431bb8485e4d670b/core/src/main/java/org/apache/flink/streaming/siddhi/router/AddRouteOperator.java#L68 Moving AddRouteOperator members in flink state for restoring after failure.
测试用例最后一个例子为什么会空指针异常 ``` @Test public void testDynamicalStreamSimplePatternMatch2() throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); DataStream input1 = env.addSource(new RandomEventSource(30).setName("event_stream_1")); DataStream controlStream = env.addSource(new SourceFunction() { @Override public void run(SourceContext sourceContext) throws...
在程序运行时，删除已有 siddhi 规则，发生报错 报错信息为： java.util.ConcurrentModificationException at java.util.HashMap$HashIterator.nextNode(HashMap.java:1442) at java.util.HashMap$KeyIterator.next(HashMap.java:1466) at org.apache.flink.streaming.siddhi.router.AddRouteOperator.handleMetadataControlEvent(AddRouteOperator.java:103) at org.apache.flink.streaming.siddhi.router.AddRouteOperator.processElement(AddRouteOperator.java:62) at org.apache.flink.streaming.runtime.io.StreamInputProcessor.processInput(StreamInputProcessor.java:202) at org.apache.flink.streaming.runtime.tasks.OneInputStreamTask.run(OneInputStreamTask.java:105) at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:300) at org.apache.flink.runtime.taskmanager.Task.run(Task.java:704) at java.lang.Thread.run(Thread.java:748)
我将WSO2 Siddhi repo下的示例 [TimeWindowSample](https://github.com/siddhi-io/siddhi/blob/master/modules/siddhi-samples/quick-start-samples/src/main/java/io/siddhi/sample/TimeWindowSample.java) 用 flink-siddhi 库改写并运行，发现和原生的示例输出结果并不相同，难道是我哪里写得不对吗？ ``` StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime); DataStream stockStream = env.fromElements( Tuple4.of("IBM", 100f, 100L, 1000L), Tuple4.of("IBM", 200f, 300L, 2000L), Tuple4.of("WSO2", 60f, 200L, 2000L),...
