THIS IS A TEST INSTANCE. ALL YOUR CHANGES WILL BE LOST!!!!
...
Code Block | ||||
---|---|---|---|---|
| ||||
Pattern.begin("start") .where((IterativeCondition) new OrderSumCondition()) .within(Time.seconds(5L)); StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(envConfiguration); DataStream<Event> input = env.fromElements( new Event(1, "barfoo", 1.0), new Event(2, "barfoo", 2.0), new Event(3, "foobar", 3.0), new Event(4, "foobar", 5.0), new Event(5, "foobar", 42.0), new Event(6, "foobar", 1.0)); DataStream<String> result = CEP.pattern(input, pattern) .inProcessingTime() .flatSelect( (p, o) -> { o.collect(String.valueOf(p.get("start").get(0).getId())); }, Types.STRING); result.print() |
...