Flink(九)CEP

这篇具有很好参考价值的文章主要介绍了Flink(九)CEP。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

1.概述

所谓 CEP,其实就是“复杂事件处理(Complex Event Processing)”的缩写;而 Flink CEP,就是 Flink 实现的一个用于复杂事件处理的库(library)。那到底什么是“复杂事件处理”呢?就是可以在事件流里,检测到特定的事件组合并进行处理,比如说“连续登录失败”,或者“订单支付超时”等等

具体的处理过程是,把事件流中的一个个简单事件,通过一定的规则匹配组合起来,这就是“复杂事件”;然后基于这些满足规则的一组组复杂事件进行转换处理,得到想要的结果进行输出

总结起来,复杂事件处理(CEP)的流程可以分成三个步骤:
(1)定义一个匹配规则,匹配规则就是模式,主要由两部分组成,每个简单事件的特征 和 简单事件之间的组合关系
(2)将匹配规则应用到事件流上,检测满足规则的复杂事件
(3)对检测到的复杂事件进行处理,得到结果进行输出

flink cep,大数据框架,flink,大数据

CEP 主要用于实时流数据的分析处理。CEP 可以帮助在复杂的、看似不相关的事件流中找出那些有意义的事件组合,进而可以接近实时地进行分析判断、输出通知信息或报警。这在企业项目的风控管理、用户画像和运维监控中,都有非常重要的应用

  • 风险控制
    当一个用户行为符合了异常行为模式,比如短时间内频繁登录并失败、大量下单却不支付(刷单),就可以向用户发送通知信息,或是进行报警提示、由人工进一步判定用户是否有违规操作的嫌疑。这样就可以有效地控制用户个人和平台的风险

  • 用户画像
    利用 CEP 可以用预先定义好的规则,对用户的行为轨迹进行实时跟踪,从而检测出具有特定行为习惯的一些用户,做出相应的用户画像。基于用户画像可以进行精准营销,即对行为匹配预定义规则的用户实时发送相应的营销推广;这与目前很多企业所做的精准推荐原理是一样的

  • 运维监控
    对于企业服务的运维管理,可以利用 CEP 灵活配置多指标、多依赖来实现更复杂的监控模式
    CEP 的应用场景非常丰富。很多大数据框架,如 Spark、Samza、Beam 等都提供了不同的CEP 解决方案,但没有专门的库(library)。而 Flink 提供了专门的 CEP 库用于复杂事件处理,可以说是目前 CEP 的最佳解决方案

2.快速入门

需要引入的依赖

<dependency>
 <groupId>org.apache.flink</groupId>
 <artifactId>flink-cep_${scala.binary.version}</artifactId>
 <version>${flink.version}</version>
</dependency>

接下来我们考虑一个具体的需求:检测用户行为,如果连续三次登录失败,就输出报警信息。很显然,这是一个复杂事件的检测处理,我们可以使用 Flink CEP 来实现。我们首先定义数据的类型。这里的用户行为不再是之前的访问事件 Event 了,所以应该单独定义一个登录事件 POJO 类。具体实现如下:

public class LoginEvent {
    // 用户id
    public String userId;
    // 用户ip地址
    public String ipAddress;
    // 用户登录成功与否
    public Boolean eventType;
    // 登录时间戳
    public Long timestamp;

  
    public LoginEvent() {}
    // 省略toString 有参构造
  
}
public class LoginDetectExample {
    public static void main(String[] args) throws Exception {
        // 创建一个表执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        env.setParallelism(1);
        env.getConfig().setAutoWatermarkInterval(100); // 100毫秒生成一次水位线
        // 1.用户登录事件
        SingleOutputStreamOperator<LoginEvent> streamOperator = env.fromElements(
                new LoginEvent("1", "192.168.10.1", false, 1000L),
                new LoginEvent("2", "192.168.10.6", true, 2000L),
                new LoginEvent("1", "192.168.10.1", false, 5000L),
                new LoginEvent("2", "192.168.10.6", false, 5000L),
                new LoginEvent("1", "192.168.10.1", false, 4000L)
        ).assignTimestampsAndWatermarks(WatermarkStrategy
                .<LoginEvent>forBoundedOutOfOrderness(Duration.ofSeconds(2)) // 延迟2秒保证数据正确
                .withTimestampAssigner(new SerializableTimestampAssigner<LoginEvent>() {
                    @Override // 时间戳的提取器
                    public long extractTimestamp(LoginEvent event, long l) {
                        return event.timestamp;
                    }
                })
        );

        // 2.定义模式
        // 2.1 模式的第一个事件是用户登陆失败
        Pattern<LoginEvent, LoginEvent> loginEventPattern = Pattern.<LoginEvent>begin("first-false")
                .where(new SimpleCondition<LoginEvent>() {
                    @Override
                    public boolean filter(LoginEvent loginEvent) throws Exception {
                        return !loginEvent.eventType; // 类型为false的则代表登陆失败
                    }
                })  // next衔接模式的第二个事件
                .next("second-false")
                .where(new SimpleCondition<LoginEvent>() {
                    @Override
                    public boolean filter(LoginEvent loginEvent) throws Exception {
                        return !loginEvent.eventType;
                    }
                }) // 以后的每个事件都用next衔接即可
                .next("third-false")
                .where(new SimpleCondition<LoginEvent>() {
                    @Override
                    public boolean filter(LoginEvent loginEvent) throws Exception {
                        return !loginEvent.eventType;
                    }
                });

        // 3.将模式应用到数据流,检测复杂事件
        PatternStream<LoginEvent> patternStream = CEP.pattern(streamOperator.keyBy(event -> event.userId), loginEventPattern);


        // 4.提取复杂事件,进行处理  select类似于map 只不过我们处理的是一组事件
        SingleOutputStreamOperator<String> warningOut = patternStream.select(new PatternSelectFunction<LoginEvent, String>() {
            @Override // 这里是一个map map的key就是我们定义的事件名称,value对应的事件列表,我们这里列表里只有一个事件,为什么是列表,因为我们定义的一个事件,它可能会重复发生
            public String select(Map<String, List<LoginEvent>> map) throws Exception {
                // 提取三次事件
                LoginEvent firstFailEvent = map.get("first-false").get(0);
                LoginEvent secondFailEvent = map.get("second-false").get(0);
                LoginEvent thirdFailEvent = map.get("third-false").get(0);


                return "用户 " + firstFailEvent.userId + "在" +
                        new Time(firstFailEvent.timestamp) + "、" +
                        new Time(secondFailEvent.timestamp) + "、" +
                        new Time(thirdFailEvent.timestamp) +
                        "已连续三次登陆失败!";
            }
        });

        // 5.打印输出
        warningOut.print();

        env.execute();
    }
}

flink cep,大数据框架,flink,大数据

3.模式API

3.1 个体模式

模式(Pattern)其实就是将一组简单事件组合成复杂事件的“匹配规则”。由于流中事件的匹配是有先后顺序的,因此一个匹配规则就可以表达成先后发生的一个个简单事件,按顺序串联组合在一起

这里的每一个简单事件并不是任意选取的,也需要有一定的条件规则;所以我们就把每个简单事件的匹配规则,叫作“个体模式”(Individual Pattern),例如上面的例子,需要匹配三次连续失败的用户,实际上就是三个个体模式

量词

个体模式后面可以跟一个“量词”,用来指定循环的次数,在上面例子中,可以在begin或者next之后接量词,注意量词的使用位置

为了更好的理解量词的含义,我们这里假设输入的4个事件为 a a a b

在 Flink CEP 中,可以使用不同的方法指定循环模式,主要有:

  • .oneOrMore()

匹配事件出现一次或多次,假设 a 是一个个体模式,a.oneOrMore()表示可以匹配 1 个或多个 a 的事件组合,匹配结果有三个 [a,a,a]、[a,a]、[a],也是说oneOrMore会以每一个匹配的事件为开头,返回最大的匹配项,第一个a,它可以匹配三个a,结束,轮到第二个a,它可以匹配两个a(第二个a本身和第三个a),结束,轮到最后一个a,只有它自己了,结束

  • .times(times)

匹配事件发生特定次数(times),例如 a.times(3)表示 aaa

  • .times(fromTimes,toTimes)

指定匹配事件出现的次数范围,最小次数为fromTimes,最大次数为toTimes。例如a.times(2, 3)可以匹配 aa,aaa

  • .greedy()

只能用在循环模式后,使当前循环模式变得“贪心”(greedy),也就是总是尽可能多地去匹配。例如 a.times(2, 4).greedy(),如果出现了连续 4 个 a,那么会直接把 aaaa 检测出来进行处理,其他任意 2 个 a 是不算匹配事件的

  • .optional()

使当前模式成为可选的,也就是说可以满足这个匹配条件,也可以不满足。对于一个个体模式 pattern 来说,后面所有可以添加的量词如下:

// 匹配事件出现 4 次
pattern.times(4);
// 匹配事件出现 4 次,或者不出现
pattern.times(4).optional();
// 匹配事件出现 2, 3 或者 4 次
pattern.times(2, 4);
// 匹配事件出现 2, 3 或者 4 次,并且尽可能多地匹配,有4次匹配4次
pattern.times(2, 4).greedy();
// 匹配事件出现 2, 3, 4 次,或者不出现
pattern.times(2, 4).optional();
// 匹配事件出现 2, 3, 4 次,或者不出现;并且尽可能多地匹配
pattern.times(2, 4).optional().greedy();
// 匹配事件出现 1 次或多次
pattern.oneOrMore();
// 匹配事件出现 1 次或多次,并且尽可能多地匹配
pattern.oneOrMore().greedy();
// 匹配事件出现 1 次或多次,或者不出现
pattern.oneOrMore().optional();
// 匹配事件出现 1 次或多次,或者不出现;并且尽可能多地匹配
pattern.oneOrMore().optional().greedy();
// 匹配事件出现 2 次或多次
pattern.timesOrMore(2);
// 匹配事件出现 2 次或多次,并且尽可能多地匹配
pattern.timesOrMore(2).greedy();
// 匹配事件出现 2 次或多次,或者不出现
pattern.timesOrMore(2).optional()
// 匹配事件出现 2 次或多次,或者不出现;并且尽可能多地匹配
pattern.timesOrMore(2).optional().greedy();

正是因为个体模式可以通过量词定义为循环模式,一个模式能够匹配到多个事件,所以之前代码中事件的检测接收才会用 Map 中的一个列表(List)来保存。而之前代码中没有定义量词,都是单例模式,所以只会匹配一个事件,每个 List 中也只有一个元素:

LoginEvent first = map.get("first").get(0);

条件 where

对于每个个体模式,匹配事件的核心在于定义匹配条件,也就是选取事件的规则。FlinkCEP 会按照这个规则对流中的事件进行筛选,判断是否接受当前的事件

  • 限定子类型

调用.subtype()方法可以为当前模式增加子类型限制条件。例如:

pattern.subtype(SubEvent.class);

这里 SubEvent 是流中数据类型 Event 的子类型。这时,只有当事件是 SubEvent 类型时,才可以满足当前模式 pattern 的匹配条件

  • 简单条件(Simple Conditions)

简单条件是最简单的匹配规则,只根据当前事件的特征来决定是否接受它。这在本质上其实就是一个 filter 操作

代码中我们为.where()方法传入一个 SimpleCondition 的实例作为参数。SimpleCondition 是表示“简单条件”的抽象类,内部有一个.filter()方法,唯一的参数就是当前事件。所以它可以当作 FilterFunction 来使用

  • 迭代条件(Iterative Conditions)

在 Flink CEP 中,提供了 IterativeCondition 抽象类。这其实是更加通用的条件表达,查看源码可以发现, .where()方法本身要求的参数类型就是 IterativeCondition;而之前 的SimpleCondition 是它的一个子类

在 IterativeCondition 中同样需要实现一个 filter()方法,不过与 SimpleCondition 中不同的是,这个方法有两个参数:除了当前事件之外,还有一个上下文 Context。调用这个上下文的.getEventsForPattern()方法,传入一个模式名称,就可以拿到这个模式中已匹配到的所有数据了

@PublicEvolving
public abstract class IterativeCondition<T> implements Function, Serializable {
    private static final long serialVersionUID = 7067817235759351255L;

    public IterativeCondition() {
    }

    public abstract boolean filter(T var1, Context<T> var2) throws Exception;

    public interface Context<T> extends TimeContext {
        Iterable<T> getEventsForPattern(String var1) throws Exception;
    }
}

下面是一个具体示例:

Pattern<LoginEvent, LoginEvent> loginEventPattern = Pattern.<LoginEvent>begin("first-false")
                .where(new SimpleCondition<LoginEvent>() {
                    @Override
                    public boolean filter(LoginEvent loginEvent) throws Exception {
                        return !loginEvent.eventType; // 类型为false的则代表登陆失败
                    }
                })  // next衔接模式的第二个事件
                .next("second-false")
                .where(new SimpleCondition<LoginEvent>() {
                    @Override
                    public boolean filter(LoginEvent loginEvent) throws Exception {
                        return !loginEvent.eventType;
                    }
                }) // 以后的每个事件都用next衔接即可
                .next("third-false")
                .where(new IterativeCondition<LoginEvent>() {
                    @Override
                    public boolean filter(LoginEvent loginEvent, Context<LoginEvent> context) throws Exception {
                        Iterable<LoginEvent> eventsForPattern = context.getEventsForPattern("second-false");
                        System.out.println(new Time(eventsForPattern.iterator().next().timestamp));
                        return !loginEvent.eventType;
                    }
                });

上面代码中,在第三个模式中,我们传入的是一个迭代条件,它调用了getEventsForPattern(“second-false”),于是获取了第二个模式中已经捕获的数据

flink cep,大数据框架,flink,大数据

  • 组合条件(Combining Conditions)

独立定义多个条件,然后在外部把它们连接起来,构成一个“组合条件”(Combining Condition)

最简单的组合条件,就是.where()后面再接一个.where()。因为前面提到过,一个条件就像是一个 filter 操作,所以每次调用.where()方法都相当于做了一次过滤,连续多次调用就表示多重过滤,最终匹配的事件自然就会同时满足所有条件。这相当于就是多个条件的“逻辑与”(AND)

而多个条件的逻辑或(OR),则可以通过.where()后加一个.or()来实现。这里的.or()方法与.where()一样,传入一个 IterativeCondition 作为参数,定义一个独立的条件;它和之前.where()定义的条件只要满足一个,当前事件就可以成功匹配

当然,子类型限定条件(subtype)也可以和其他条件结合起来,成为组合条件,如下所示:

pattern.subtype(SubEvent.class)
.where(new SimpleCondition<SubEvent>() {
 @Override
 public boolean filter(SubEvent value) {
 return ... // some condition
 }
});

这里可以看到,SimpleCondition 的泛型参数也变成了 SubEvent,所以匹配出的事件就既满足子类型限制,又符合过滤筛选的简单条件;这也是一个逻辑与的关系

  • 终止条件(Stop Conditions)

对于循环模式而言,还可以指定一个“终止条件”(Stop Condition),表示遇到某个特定事件时当前模式就不再继续循环匹配了

终 止 条 件 的 定 义 是 通 过 调 用 模 式 对 象 的 .until() 方 法 来 实 现 的 , 同 样 传 入 一 个IterativeCondition 作为参数。需要注意的是,终止条件只与 oneOrMore() 或 者oneOrMore().optional()结合使用。因为在这种循环模式下,我们不知道后面还有没有事件可以匹配,只好把之前匹配的事件作为状态缓存起来继续等待,这等待无穷无尽;如果一直等下去,缓存的状态越来越多,最终会耗尽内存。所以这种循环模式必须有个终点,当.until()指定的条件满足时,循环终止,这样就可以清空状态释放内存了

3.2 组合模式

有了定义好的个体模式,就可以尝试按一定的顺序把它们连接起来,定义一个完整的复杂事件匹配规则了。这种将多个个体模式组合起来的完整模式,就叫作“组合模式”(Combining Pattern),为了跟个体模式区分有时也叫作“模式序列”(Pattern Sequence)

组合模式就是一个“模式序列”,是用诸如 begin、next、followedBy 等表示先后顺序的“连接词”将个体模式串连起来得到的。在这样的语法调用中,每个事件匹配的条件是什么、各个事件之间谁先谁后、近邻关系如何都定义得一目了然。每一个“连接词”方法调用之后,得到的都仍然是一个 Pattern 的对象;所以从 Java 对象的角度看,组合模式与个体模式是一样的,都是 Pattern

1. 初始模式(Initial Pattern)

所有的组合模式,都必须以一个“初始模式”开头;而初始模式必须通过调用 Pattern 的静态方法.begin()来创建。如下所示:

Pattern<Event, ?> start = Pattern.<Event>begin("start");

这里我们调用 Pattern 的.begin()方法创建了一个初始模式。传入的 String 类型的参数就是模式的名称;而 begin 方法需要传入一个类型参数,这就是模式要检测流中事件的基本类型,这里我们定义为 Event。调用的结果返回一个 Pattern 的对象实例。Pattern 有两个泛型参数,第一个就是检测事件的基本类型 Event,跟 begin 指定的类型一致;第二个则是当前模式里事件的子类型,由子类型限制条件指定。我们这里用类型通配符(?)代替,就可以从上下文直接推断了

2. 近邻条件(Contiguity Conditions)

在初始模式之后,我们就可以按照复杂事件的顺序追加模式,组合成模式序列了。模式之间的组合是通过一些“连接词”方法实现的,这些连接词指明了先后事件之间有着怎样的近邻关系,这就是所谓的“近邻条件”(Contiguity Conditions,也叫“连续性条件”)

Flink CEP 中提供了三种近邻关系:

  • 严格近邻(Strict Contiguity)

匹配的事件严格地按顺序一个接一个出现,中间不会有任何其他事件
代码中对应的就是 Pattern 的.next()方法,名称上就能看出来,“下一个”自然就是紧挨着的

flink cep,大数据框架,flink,大数据

  • 宽松近邻(Relaxed Contiguity)

宽松近邻只关心事件发生的顺序,而放宽了对匹配事件的“距离”要求,也就是说两个匹配的事件之间可以有其他不匹配的事件出现。代码中对应.followedBy()方法,很明显这表示“跟在后面”就可以,不需要紧紧相邻

flink cep,大数据框架,flink,大数据

  • 非确定性宽松近邻(Non-Deterministic Relaxed Contiguity)

也就是从全局找所有符合匹配模式的事件队列,下图我们找圆 + 三角,代码中对应.followedByAny()方法

flink cep,大数据框架,flink,大数据

3. 其他限制条件

除了上面提到的 next()、followedBy()、followedByAny()可以分别表示三种近邻条件,我们还可以用否定的“连接词”来组合个体模式。主要包括:

  • .notNext()

表示前一个模式匹配到的事件后面,不能紧跟着某种事件

  • .notFollowedBy()

表示前一个模式匹配到的事件后面,不会出现某种事件

这里需要注意,由于notFollowedBy()是没有严格限定的;流数据不停地到来,我们永远不能保证之后“不会出现某种事件”。所以一个模式序列不能以 notFollowedBy()结尾,这个限定条件主要用来表示“两个事件中间不会出现某种事件”

  • .within()

方法传入一个时间参数,这是模式序列中第一个事件到最后一个事件之间的最大时间间隔,只有在这期间成功匹配的复杂事件才是有效的。一个模式序列中只能有一个时间限制,调用.within()的位置不限;如果多次调用则会以最小的那个时间间隔为准

下面是模式序列中所有限制条件在代码中的定义:

// 严格近邻条件
Pattern<Event, ?> strict = start.next("middle").where(...);
// 宽松近邻条件
Pattern<Event, ?> relaxed = start.followedBy("middle").where(...);
// 非确定性宽松近邻条件
Pattern<Event, ?> nonDetermin = 
start.followedByAny("middle").where(...);
// 不能严格近邻条件
Pattern<Event, ?> strictNot = start.notNext("not").where(...);
// 不能宽松近邻条件
Pattern<Event, ?> relaxedNot = start.notFollowedBy("not").where(...);
// 时间限制条件
middle.within(Time.seconds(10));

4. 循环模式中的近邻条件

在循环模式中,近邻关系同样有三种:严格近邻、宽松近邻以及非确定性宽松近邻。对于定义了量词(如 oneOrMore()、times())的循环模式,默认内部采用的是宽松近邻。也就是说,当循环匹配多个事件时,它们中间是可以有其他不匹配事件的;相当于用单例模式分别定义、再用 followedBy()连接起来

  • .consecutive()

循环模式中的匹配事件增加严格的近邻条件,保证所有匹配事件是严格连续的。也就是说,一旦中间出现了不匹配的事件,当前循环检测就会终止。这起到的效果跟模式序列中的next()一样,需要与循环量词 times()、oneOrMore()配合使用

  • .allowCombinations()

除严格近邻外,也可以为循环模式中的事件指定非确定性宽松近邻条件,表示可以重复使用 已 经 匹 配 的 事 件。 这 需 要 调 用 .allowCombinations() 方 法 来 实 现 , 实 现 的 效 果与.followedByAny()相同

flink cep,大数据框架,flink,大数据

3.3 匹配后跳过策略

在 Flink CEP 中,由于有循环模式和非确定性宽松近邻的存在,同一个事件有可能会重复利用,被分配到不同的匹配结果中。这样会导致匹配结果规模增大,有时会显得非常冗余。当然,非确定性宽松近邻条件,本来就是为了放宽限制、扩充匹配结果而设计的;我们主要是针对循环模式来考虑匹配结果的精简

之前已经讲过,如果对循环模式增加了.greedy()的限制,那么就会“尽可能多地”匹配事件,这样就可以砍掉那些子集上的匹配了。不过这种方式还是略显简单粗暴,如果我们想要精确控制事件的匹配应该跳过哪些情况,那就需要制定另外的策略了

在 Flink CEP 中,提供了模式的“匹配后跳过策略”(After Match Skip Strategy),专门用来精准控制循环模式的匹配结果。这个策略可以在 Pattern 的初始模式定义中,作为 begin()的第二个参数传入:

Pattern.begin("start", AfterMatchSkipStrategy.noSkip())
.where(...)

flink cep,大数据框架,flink,大数据

具体的跳过策略有5种

public abstract class AfterMatchSkipStrategy implements Serializable {
    private static final long serialVersionUID = -4048930333619068531L;

    public static SkipToFirstStrategy skipToFirst(String patternName) {
        return new SkipToFirstStrategy(patternName, false);
    }

    public static SkipToLastStrategy skipToLast(String patternName) {
        return new SkipToLastStrategy(patternName, false);
    }

    public static SkipPastLastStrategy skipPastLastEvent() {
        return SkipPastLastStrategy.INSTANCE;
    }

    public static AfterMatchSkipStrategy skipToNext() {
        return SkipToNextStrategy.INSTANCE;
    }

    public static NoSkipStrategy noSkip() {
        return NoSkipStrategy.INSTANCE;
    }

}

案例

输入为 a a a b b

Pattern<Word, Word> wordPattern = Pattern.<Word>begin("first").oneOrMore()
                .where(new SimpleCondition<Word>() {
                    @Override
                    public boolean filter(Word word) throws Exception {
                        return word.ch.equals('a');
                    }
                }).next("second").oneOrMore()
                .where(new SimpleCondition<Word>() {
                    @Override
                    public boolean filter(Word word) throws Exception {
                        return word.ch.equals('b');
                    }
                });

  • 不跳过(NO_SKIP)

代码调用 AfterMatchSkipStrategy.noSkip()。这是默认策略,所有可能的匹配都会输出

返回 [a,a,a,b] 、[a,a,b] 、[a,b]、 [a,a,a,b,b] 、 [a,a,b,b] 、[a,b,b]

  • 跳至下一个(SKIP_TO_NEXT)

代码调用 AfterMatchSkipStrategy.skipToNext(),找到一个匹配项,跳至下一个元素

返回 [a,a,a,b] 、[a,a,b] 、[a,b]

  • 跳过所有子匹配(SKIP_PAST_LAST_EVENT)

代码调用 AfterMatchSkipStrategy.skipPastLastEvent(),只返回第一个匹配也即跳过所有子匹配,这是最为精简的跳过策略

返回[a,a,a,b]

  • 跳至第一个(SKIP_TO_FIRST[])

代码调用 AfterMatchSkipStrategy.skipToFirst(“second”),这里传入一个参数,指明跳至哪个模式的第一个匹配事件

返回 [a,a,a,b] 、[a,a,b] 、[a,b] ,second的一个匹配为b

  • 跳至最后一个(SKIP_TO_LAST[])

代码调用 AfterMatchSkipStrategy.skipToLast(“second”),同样传入一个参数,指明跳至哪个模式的最后一个匹配事件

4.模式的检测处理

4.1 模式应用到数据流

将模式应用到事件流上的代码非常简单,只要调用 CEP 类的静态方法.pattern(),将数据流(DataStream)和模式(Pattern)作为两个参数传入就可以了。最终得到的是一个 PatternStream:

DataStream<Event> inputStream = ...
Pattern<Event, ?> pattern = ...
PatternStream<Event> patternStream = CEP.pattern(inputStream, pattern);

这里的 DataStream,也可以通过 keyBy 进行按键分区得到 KeyedStream,接下来对复杂事件的检测就会针对不同的 key 单独进行了

模式中定义的复杂事件,发生是有先后顺序的,这里“先后”的判断标准取决于具体的时间语义。默认情况下采用事件时间语义,那么事件会以各自的时间戳进行排序;如果是处理时间语义,那么所谓先后就是数据到达的顺序。对于时间戳相同或是同时到达的事件,我们还可以在 CEP.pattern()中传入一个比较器作为第三个参数,用来进行更精确的排序:

// 可选的事件比较器
EventComparator<Event> comparator = ... 
PatternStream<Event> patternStream = CEP.pattern(input, pattern, comparator);

得到 PatternStream 后,接下来要做的就是对匹配事件的检测处理了

4.2 处理匹配事件

基于 PatternStream 可以调用一些转换方法,对匹配的复杂事件进行检测和处理,并最终得到一个正常的 DataStream。这个转换的过程与窗口的处理类似:将模式应用到流上得到PatternStream,就像在流上添加窗口分配器得到 WindowedStream;而之后的转换操作,就像定义具体处理操作的窗口函数,对收集到的数据进行分析计算,得到结果进行输出,最后回到DataStream 的类型来

PatternStream 的转换操作主要可以分成两种:简单便捷的选择提取(select)操作,和更加通用、更加强大的处理(process)操作。与 DataStream 的转换类似,具体实现也是在调用API 时传入一个函数类:选择操作传入的是一个 PatternSelectFunction,处理操作传入的则是一个 PatternProcessFunction

1. 匹配事件的选择提取(select)

flink cep,大数据框架,flink,大数据

处理匹配事件最简单的方式,就是从 PatternStream 中直接把匹配的复杂事件提取出来,包装成想要的信息输出,这个操作就是“选择”(select)

  • PatternSelectFunction

代码中基于 PatternStream 直接调用.select()方法,传入一个 PatternSelectFunction 作为参数

public interface PatternSelectFunction<IN, OUT> extends Function, Serializable {
    OUT select(Map<String, List<IN>> var1) throws Exception;
}

它会将检测到的匹配事件保存在一个 Map 里,对应的 key 就是这些事件的名称。这里的“事件名称”就对应着在模式中定义的每个个体模式的名称;而个体模式可以是循环模式,一个名称会对应多个事件,所以最终保存
在 Map 里的 value 就是一个事件的列表(List)

2.flatSelect

flink cep,大数据框架,flink,大数据

  • PatternFlatSelectFunction

PatternStream 还有一个类似的方法是.flatSelect(),传入的参数是一个PatternFlatSelectFunction。从名字上就能看出,这是 PatternSelectFunction 的“扁平化”版本;内部需要实现一个 flatSelect()方法,它与之前 select()的不同就在于没有返回值,而是多了一个收集器(Collector)参数 out,通过调用 out.collet()方法就可以实现多次发送输出数据了

public interface PatternFlatSelectFunction<IN, OUT> extends Function, Serializable {
    void flatSelect(Map<String, List<IN>> var1, Collector<OUT> var2) throws Exception;
}

3.process

flink cep,大数据框架,flink,大数据

@PublicEvolving
public abstract class PatternProcessFunction<IN, OUT> extends AbstractRichFunction {
    public PatternProcessFunction() {
    }

    public abstract void processMatch(Map<String, List<IN>> var1, Context var2, Collector<OUT> var3) throws Exception;

    public interface Context extends TimeContext {
        <X> void output(OutputTag<X> var1, X var2);
    }
}
4.3 处理超时事件

复杂事件的检测结果一般只有两种:要么匹配,要么不匹配。检测处理的过程具体如下:
(1)如果当前事件符合模式匹配的条件,就接受该事件,保存到对应的 Map 中
(2)如果在模式序列定义中,当前事件后面还应该有其他事件,就继续读取事件流进行检测;如果模式序列的定义已经全部满足,那么就成功检测到了一组匹配的复杂事件,调用PatternProcessFunction 的 processMatch()方法进行处理
(3)如果当前事件不符合模式匹配的条件,就丢弃该事件
(4)如果当前事件破坏了模式序列中定义的限制条件,比如不满足严格近邻要求,那么当前已检测的一组部分匹配事件都被丢弃,重新开始检测

不过在有时间限制的情况下,需要考虑的问题会有一点特别。比如我们用.within()指定了模式检测的时间间隔,超出这个时间当前这组检测就应该失败了。然而这种“超时失败”跟真正的“匹配失败”不同,它其实是一种“部分成功匹配”;因为只有在开头能够正常匹配的前提下,没有等到后续的匹配事件才会超时。所以往往不应该直接丢弃,而是要输出一个提示或报警信息。这就要求我们有能力捕获并处理超时事件

使用 PatternProcessFunction 的侧输出流

在 Flink CEP 中 , 提 供 了 一 个 专 门 捕 捉 超 时 的 部 分 匹 配 事 件 的 接 口 , 叫 作TimedOutPartialMatchHandler。这个接口需要实现一个 processTimedOutMatch()方法,可以将超时的、已检测到的部分匹配事件放在一个 Map 中,作为方法的第一个参数;方法的第二个参数则是 PatternProcessFunction 的上下文 Context。所以这个接口必须与 PatternProcessFunction结合使用,对处理结果的输出则需要利用侧输出流来进行。代码中的调用方式如下:

class MyPatternProcessFunction extends PatternProcessFunction<Event, String> implements TimedOutPartialMatchHandler<Event> {
 // 正常匹配事件的处理
@Override
 public void processMatch(Map<String, List<Event>> match, Context ctx, 
Collector<String> out) throws Exception{
 ...
 }
 // 超时部分匹配事件的处理
 @Override
 public void processTimedOutMatch(Map<String, List<Event>> match, Context ctx) 
throws Exception{
 Event startEvent = match.get("start").get(0);
 OutputTag<Event> outputTag = new OutputTag<Event>("time-out"){};
 ctx.output(outputTag, startEvent);
 }
}

我们在 processTimedOutMatch()方法中定义了一个输出标签(OutputTag)。调用 ctx.output()方法,就可以将超时的部分匹配事件输出到标签所标识的侧输出流了

案例,检测用户订单是否超时

public class TimeOutTest {
    public static void main(String[] args) throws Exception {
        // 创建一个表执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        env.setParallelism(1);
        env.getConfig().setAutoWatermarkInterval(100); // 100毫秒生成一次水位线
        // 1.用户行为事件 检测用户订单是否超时 我们假设订单从创建到支付只有20分钟,并且可以中途修改,但是修改后不会重新计时
        SingleOutputStreamOperator<OrderEvent> streamOperator = env.fromElements(
                new OrderEvent("1", "101", "create", 1000L),
                new OrderEvent("2", "102", "create", 20000L),
                new OrderEvent("1", "101", "update", 10 * 60 * 1000L),
                new OrderEvent("1", "101", "pay", 15 * 60 * 1000L)
        ).assignTimestampsAndWatermarks(WatermarkStrategy
                .<OrderEvent>forBoundedOutOfOrderness(Duration.ofSeconds(2)) // 延迟2秒保证数据正确
                .withTimestampAssigner(new SerializableTimestampAssigner<OrderEvent>() {
                    @Override // 时间戳的提取器
                    public long extractTimestamp(OrderEvent event, long l) {
                        return event.timestamp;
                    }
                })
        );
        // 2.定义模式
        Pattern<OrderEvent, OrderEvent> pattern = Pattern.<OrderEvent>begin("first")
                .where(new SimpleCondition<OrderEvent>() {
                    @Override
                    public boolean filter(OrderEvent orderEvent) throws Exception {
                        return orderEvent.action.equals("create");
                    }
                })
                .followedBy("pay")  // 因为中途可以修改订单,所以是非严格近邻
                .where(new SimpleCondition<OrderEvent>() {
                    @Override
                    public boolean filter(OrderEvent orderEvent) throws Exception {
                        return orderEvent.action.equals("pay");
                    }
                })
                .within(Time.minutes(15));


        // 3.将模式应用到数据流上
        PatternStream<OrderEvent> patternStream = CEP.pattern(streamOperator.keyBy(event -> event.orderId), pattern);

        // 4侧输出流 定义侧输出流标签
        OutputTag<String> outputTag = new OutputTag<String>("timeout") {};

        // 5.处理匹配到的数据
        SingleOutputStreamOperator<String> res = patternStream.process(new MyPatternProcessFunction(outputTag));


        res.print("payed: =>");
        res.getSideOutput(outputTag).print("timeout: =>");

        env.execute();


    }

    public static class MyPatternProcessFunction extends PatternProcessFunction<OrderEvent,String> implements TimedOutPartialMatchHandler<OrderEvent>{
        public OutputTag<String> outputTag;
        public MyPatternProcessFunction(OutputTag<String> outputTag){
            this.outputTag = outputTag;
        }

        // 获取正常的匹配事件
        @Override
        public void processMatch(Map<String, List<OrderEvent>> map, Context context, Collector<String> collector) throws Exception {
            OrderEvent orderEvent = map.get("pay").get(0);

            collector.collect("用户 "+ orderEvent.userId + "的订单=>" + orderEvent.orderId + "正常支付!");
        }
        // 处理超时的事件
        @Override
        public void processTimedOutMatch(Map<String, List<OrderEvent>> map, Context context) throws Exception {
            OrderEvent orderEvent = map.get("first").get(0);
            context.output(outputTag,"用户 "+ orderEvent.userId + "的订单=>" + orderEvent.orderId + "超时!");
        }
    }

}

flink cep,大数据框架,flink,大数据文章来源地址https://www.toymoban.com/news/detail-795184.html

到了这里,关于Flink(九)CEP的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • Flink-CEP 实战教程

      所谓CEP,其实就是“ 复杂事件处理(Complex Event Processing) ”的缩写;而 Flink CEP,就是 Flink 实现的一个用于复杂事件处理的库(library)。   那到底什么是“复杂事件处理”呢?就是可以在事件流里,检测到特定的事件组合并进行处理,比如说“连续登录失败”,或者

    2024年02月02日
    浏览(63)
  • Flink CEP(三)pattern动态更新

    目录 1.实现分析 2.代码实现 3.测试验证 4.源码地址          线上运行的CEP中肯定经常遇到规则变更的情况,如果每次变更时都将任务重启、重新发布是非常不优雅的。尤其在营销或者风控这种对实时性要求比较高的场景,如果规则窗口过长(一两个星期),状态过大,就

    2024年02月13日
    浏览(44)
  • Flink CEP(Complex Event Processing)库

    复杂事件处理(Complex Event Processing,CEP)是一种用于在流式数据中识别和处理复杂事件模式的技术。Apache Flink 作为一个流式处理框架,也可以用于实现复杂事件处理。下面是 Flink 中实现复杂事件处理的一般原理: 事件流输入: 首先,Flink 接收外部的事件流作为输入。这些事

    2024年02月13日
    浏览(37)
  • Flink CEP(三)pattern动态更新(附源码)

    目录 1.实现分析 2.代码实现 3.测试验证 4.源码地址          线上运行的CEP中肯定经常遇到规则变更的情况,如果每次变更时都将任务重启、重新发布是非常不优雅的。尤其在营销或者风控这种对实时性要求比较高的场景,如果规则窗口过长(一两个星期),状态过大,就

    2024年02月04日
    浏览(47)
  • Flink 内容分享(二十七):Hadoop vs Spark vs Flink——大数据框架比较

    大数据开发离不开各种框架,我们通过学习 Apache Hadoop、Spark 和 Flink 之间的特征比较,可以从侧面了解要学习的内容。众所周知,Hadoop vs Spark vs Flink是快速占领 IT 市场的三大大数据技术,大数据岗位几乎都是围绕它们展开。 本文,将详细介绍三种框架之间的区别。 Hadoop:为

    2024年02月01日
    浏览(69)
  • Flink CDC 新一代数据集成框架

    前言: 主要讲解了技术原理,入门与生产实践,主要功能:全增量一体化数据集成、实时数据入库入仓、最详细的教程。Flink CDC 是Apache Flink的一个重要组件,主要使用了CDC技术从各种数据库中获取变更流并接入到Flink中,Apache Flink作为一款非常优秀的流处理引擎,其SQL API又

    2024年02月13日
    浏览(62)
  • 数据流处理框架Flink与Kafka

    在大数据时代,数据流处理技术已经成为了一种重要的技术手段,用于处理和分析大量实时数据。Apache Flink和Apache Kafka是两个非常重要的开源项目,它们在数据流处理领域具有广泛的应用。本文将深入探讨Flink和Kafka的关系以及它们在数据流处理中的应用,并提供一些最佳实践

    2024年04月23日
    浏览(41)
  • 大数据学习之Flink、比较不同框架的容错机制

     第一章、Flink的容错机制 第二章、Flink核心组件和工作原理 第三章、Flink的恢复策略 第四章、Flink容错机制的注意事项 第五章、Flink的容错机制与其他框架的容错机制相比较 目录 第五章、Flink的容错机制与其他框架的容错机制相比较 Ⅰ、Flink的容错机制与其他框架的容错机

    2024年01月23日
    浏览(47)
  • 大数据Flink(四十九):框架版本介绍和编程语言选择

    文章目录 框架版本介绍和编程语言选择 一、框架版本介绍 二、编程语言选择

    2024年02月15日
    浏览(53)
  • Flink CDC 3.0 正式发布,详细解读新一代实时数据集成框架

    Flink CDC 是基于数据库日志 CDC(Change Data Capture)技术的实时数据集成框架,支持了全增量一体化、无锁读取、并行读取、表结构变更自动同步、分布式架构等高级特性。配合 Flink 优秀的管道能力和丰富的上下游生态,Flink CDC 可以高效实现海量数据的实时集成。Flink CDC 社区发

    2024年02月04日
    浏览(55)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包