Flink难点：彻底明白CEP2，条件分类

2019-6-11 10:32| 发布者: admin| 查看: 1| 评论: 0|原作者: pig2

摘要: 问题导读 1.对于接受事件该如何实现过滤？ 2.CEP都有哪些条件？ 3.迭代条件和简单条件有什么区别？ 4.如何组合条件？接上篇 Flink难点CEP1：什么是CEP以及量词的含义 http://www.aboutyun.com/forum.p ...

问题导读

1.对于接受事件该如何实现过滤？
2.CEP都有哪些条件？
3.迭代条件和简单条件有什么区别？
4.如何组合条件？

接上篇
Flink难点CEP1：什么是CEP以及量词的含义
http://www.aboutyun.com/forum.php?mod=viewthread&tid=27142

条件
对于每个模式，可以指定接受事件必须满足的条件，用来“传入”到模式中，例如其值应大于5，或大于先前接受的事件的平均值。可以通过pattern.where（），pattern.or（）或pattern.until（）方法指定事件属性的条件。条件可以分为IterativeConditions（迭代条件）和SimpleConditions（简单条件）。

迭代条件：这是最常见的条件类型。这是可以如何指定一个条件，该条件基于先前接受的事件的属性或其子集的统计信息来接受后续事件。

下面是迭代条件的代码，如果名称以“foo”开头，则接受名为“middle”的模式的下一个事件，并且如果该模式的先前接受的事件的价格总和加上当前的价格事件不超过5.0的值。迭代条件是非常有用的，尤其是与循环模式组合，例如， oneOrMore().。

[Scala] 纯文本查看 复制代码

middle.oneOrMore()

    .subtype(classOf[SubEvent])

    .where(

        (value, ctx) => {

            lazy val sum = ctx.getEventsForPattern("middle").map(_.getPrice).sum

            value.getName.startsWith("foo") && sum + value.getPrice < 5.0

        }

    )

[Java] 纯文本查看 复制代码

middle.oneOrMore()

    .subtype(SubEvent.class)

    .where(new IterativeCondition<SubEvent>() {

        @Override

        public boolean filter(SubEvent value, Context<SubEvent> ctx) throws Exception {

            if (!value.getName().startsWith("foo")) {

                return false;

            }

    

            double sum = value.getPrice();

            for (Event event : ctx.getEventsForPattern("middle")) {

                sum += event.getPrice();

            }

            return Double.compare(sum, 5.0) < 0;

        }

    });

注意对ctx.getEventsForPattern（...）的调用将查找给定潜在匹配项的所有先前接受的事件。此操作的成本可能会有所不同，因此在实施的条件时，请尽量减少其使用。

所描述的上下文也提供了对事件时间特征的访问。有关更多信息，请参阅时间上下文（Time context.）

简单条件：这种类型的条件扩展了前面提到的IterativeCondition类，并且仅根据事件本身的属性决定是否接受事件。

[Scala] 纯文本查看 复制代码

start.where(event => event.getName.startsWith("foo"))

[Java] 纯文本查看 复制代码

start.where(new SimpleCondition<Event>() {

    @Override

    public boolean filter(Event value) {

        return value.getName().startsWith("foo");

    }

});

最后，还可以通过pattern.subtype（subClass）方法将接受事件的类型限制为初始事件类型（此处为Event）的子类型。

[Scala] 纯文本查看 复制代码

start.subtype(classOf[SubEvent]).where(subEvent => ... /* some condition */)

[Java] 纯文本查看 复制代码

start.subtype(SubEvent.class).where(new SimpleCondition<SubEvent>() {

    @Override

    public boolean filter(SubEvent value) {

        return ... // some condition

    }

});

组合条件：如上所示，您可以将子类型条件与其他条件组合使用。这适用于所有条件。您可以通过顺序调用where（）来任意组合条件。最终结果将是各个条件的结果的逻辑AND。要使用OR组合条件，可以使用or（）方法，如下所示。

[Scala] 纯文本查看 复制代码

pattern.where(event => ... /* some condition */).or(event => ... /* or condition */)

[Java] 纯文本查看 复制代码

pattern.where(new SimpleCondition<Event>() {

    @Override

    public boolean filter(Event value) {

        return ... // some condition

    }

}).or(new SimpleCondition<Event>() {

    @Override

    public boolean filter(Event value) {

        return ... // or condition

    }

});

停止条件：在循环模式（oneOrMore（）和oneOrMore（）。optional（））的情况下，还可以指定停止条件，例如：接受值大于5的事件，直到值的总和小于50。

为了更好地理解它，请看下面的示例。特定

示例：

1.模式如“（a +until b）”（one or more "a" until "b"）

2.一系列传入事件“a1”“c”“a2”“b”“a3”

3.将输出结果：{a1 a2} {a1} {a2} {a3}。

如所见，由于停止条件，未返回{a1 a2 a3}或{a2 a3}。

最新经典文章，欢迎关注公众号