Flink 使用之 CEP

Flink 使用介绍相关文档目录

Flink 使用介绍相关文档目录

什么是CEP

CEP的全称为Complex Event Processing，中文翻译为复杂事件处理。光看字面意思解释还是很难理解。究竟何为“复杂事件”？通常我们使用Flink处理数据流的时候，只是对每个到来的元素感兴趣，不关注元素之间的关系。即便是有也仅仅是使用有状态算子而已。现在有一种需求，我们需要关注并捕获一系列有特定规律的事件，比方说用户登录，转帐，然后退出（ABC事件连续发生），或者是比如机房连续10次测温均高于50度（A{10,}），我们采用传统方式写Flink程序就比较困难。这时候就轮到Flink CEP大显身手了。

下面为大家讲解下Flink CEP的使用方式

使用Java编写代码需要引入： 

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-cep_2.11</artifactId>
    <version>1.13.2</version>
</dependency>

其中version对应Flink版本。


使用Scala则需要引入：


<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-cep-scala_2.11</artifactId>
    <version>1.13.2</version>
</dependency>

下面的代码均以Java为准。


Pattern API


Pattern即我们在第一节提到的用户登录，转帐，然后退出或者机房连续10次测温均高于50度。和正则表达式的Pattern概念很类似。Pattern就是用户定义连续一系列事件应该具有的特征的编程接口。


我们以一个简单的例子开始：编写一个连续10次以上机房测温高于50度的Pattern。


Pattern<Row, Row> pattern = Pattern.<Row>begin("high").where(new SimpleCondition<Row>() {
    @Override
    public boolean filter(Row row) throws Exception {
        int temp = (int) row.getField("temperature");
        return temp >= 50;
}).timesOrMore(10).consecutive();

首先我们调用Pattern的begin方法，标记Pattern的开始，需要为开始的Pattern指定一个名称。一个Pattern可以分为多段，每段都具有自己的名称，在后面捕获匹配数据流的元素时候需要用到。begin方法需要指定泛型，即数据源的数据类型。然后跟随的是一个where条件。where方法接收一个SimpleCondition内部类，对用户数据进行解析和条件判断，符合条件的元素返回true。接下来是指定满足条件的元素的个数，这里使用了timesOrMore(n)，含义为n次或n次以上。我们注意到后面还有一个consecutive，意味着timesOrMore(n)必须是连续的，中间不能够穿插有其他元素。


我们再举一个用户登录，取款然后退出的例子：


// 分别定义用户的登陆事件，提款事件和登出事件
class UserEvent {
class LoginEvent extends UserEvent {
class WithdrawEvent extends UserEvent {
class LogoutEvent extends UserEvent {
// 下面是Pattern
Pattern
        .<UserEvent>begin("login")
        .subtype(LoginEvent.class)
        .next("withdraw")
        .subtype(WithdrawEvent.class)
        .timesOrMore(1)
        .consecutive()
        .next("logout")
        .subtype(LogoutEvent.class);

这里定义的Pattern为AB+C，包含三种事件分别命名为"login"，"withdraw"和"logout"。使用subtype指定满足条件的数据类型。即如果连续到来3个元素分别为LoginEvent，WithdrawEvent和LogoutEvent的实例，这3个元素会被捕获。


实例已经讲完了大家对Pattern的使用应该有了初步了解。下面开始逐个分析Pattern API更为详细的配置。


Pattern 组合方式配置


定义组合配置主要是如下3个方法：


  • begin：匹配模版的开始。
    • 

  • next：严格匹配，A next B含义为A后必须紧跟着B。
    • 

  • followedBy：非严格匹配，不要求非得紧跟着，中间可以穿插其他元素。例如A followedBy B不仅能匹配A B，还能匹配A C B。C未能匹配会被忽略掉。仅仅忽略未被匹配的元素。
    • 

  • followedByAny：非严格匹配，比followedBy更加宽松，甚至能忽略掉可被匹配的元素，例如A followedByAny B去匹配A C B1 B2可以匹配到A B1（C被忽略）和A B2（尽管B1符合条件，但是也能被忽略掉，这是followedByAny和followedBy的不同之处）。如果是A followedBy B仅仅能匹配出A B1。
    • 

  • notNext：next的否定形式
    • 

  • notFollowedBy：followedBy的否定形式，注意notFollowedBy不能用在Pattern的结尾。
    • 

    指定重复次数
    

    重复次数可以指定0次，1次，n次，n次到m次，n次以上等等。也可以限定这n次之间是否可以穿插其他事件。
    

    配置的方法为：
    

  • times(n)：n次
    • 

  • times(n, m)：n次到m次
    • 

  • optional()：带上这个允许出现0次
    • 

  • oneOrMore()：1次或多次
    • 

  • timesOrMore(n)：n次或多次
    • 

  • greedy()：表示尽可能匹配次数多的。例如到来的元素为 A X X B。其中X既满足条件A又满足条件B。对于Pattern A+ B，可以匹配到A X，A X X，X X B，X B和 A X X B。但是对于A (greedy)+ B，Flink会尽可能的多匹配A条件，故只会匹配到A X X B。
    • 

  • consecutive()：要求Pattern必须连续。
    • 

  • allowCombinations()：和consecutive相反，不要求连续。
    • 

  • where()：参数为一个内部类，非常类似于filter算子，可编写自定义条件表达式。
    • 

  • subtype()：指定匹配元素的类型。
    • 

    匹配后跳过策略
    

    我们经常遇到一个元素可以被成功匹配多次的情况。在实际应用中，一个元素究竟可以被如何匹配，这种行为可以通过匹配后跳过策略来指定。
    

    匹配后跳过策略有如下5种：
    

  • NO_SKIP: 不跳过，匹配所有的可能性。
    • 

  • SKIP_TO_NEXT: 从匹配成功的事件序列中的第一个事件的下一个事件开始进行下一次匹配。
    • 

  • SKIP_PAST_LAST_EVENT: 从匹配成功的事件序列中的最后一个事件的下一个事件开始进行下一次匹配。
    • 

  • SKIP_TO_FIRST(patternName): 从匹配成功的事件序列中第一个对应于patternName的事件开始进行下一次匹配。
    • 

  • SKIP_TO_LAST(patternName): 从匹配成功的事件序列中最后一个对应于patternName的事件开始进行下一次匹配。
    • 

    匹配后跳过策略在定义组合配置的时候指出：
    

    Pattern.begin("patternName", skipStrategy);

    skipStrategy通过如下方式创建：
    

    AfterMatchSkipStrategy.noSkip();
AfterMatchSkipStrategy.skipToNext();
AfterMatchSkipStrategy.skipPastLastEvent();
AfterMatchSkipStrategy.skipToFirst(patternName);
AfterMatchSkipStrategy.skipToLast(patternName);

    除了从数据上约束之外，Flink CEP还支持从时间维度来指定Pattern。
    

  • within(时间段)：这些满足条件的一连串元素必须发生在指定时间段之内。
    • 

    创建PatternStream
    

    通过CEP.pattern方法，关联数据源DataStream和上面章节我们创建出的Pattern。
    

    PatternStream<Event> patternStream = CEP.pattern(input, pattern);
// 或者我们指定comparator，不是很常用
PatternStream<Event> patternStream = CEP.pattern(input, pattern, comparator);

    其中第二个方法传入的Comparator可以自定义元素比较的方法，用于当元素的Event Time相同的时候来判断先后顺序。例如我们编写一个自定义的Row类型比较逻辑：
    

    new EventComparator<Row>() {
    @Override
    public int compare(Row o1, Row o2) {
        // 自定义比较逻辑在此
        return 0;

    需要注意的是，Flink 1.12版本之后CEP的PatternStream默认使用Event Time。如果业务使用的是Processing Time，必须要明确配置。
    

    PatternStream<Event> patternStream = CEP.pattern(input, pattern).inProcessingTime();

    从PatternStream中选出捕获的元素
    

    select方法接收一个PatternSelectFunction类型参数，需要用户实现这个接口，编写自己的处理逻辑。接口如下所示：
    

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

    接收到的参数类型为Map<String, List<IN>>，其中map的key为Pattern组合方式中指定的pattern名称，value为匹配到的一系列元素组成的集合。将处理过后的数据通过return返回即可。
    

    还可以使用process函数：
    

    CEP.pattern(...).process(new PatternProcessFunction<IN, OUT>() {
    @Override
    public void processMatch(Map<String, List<IN>> map, Context context, Collector<OUT> collector) throws Exception {
        // ...

    这里的参数类型和PatternSelectFunction类似，但是处理过后的数据不能通过return返回，需要使用collector收集。
    

    对于Pattern可以匹配到，但是超时的元素（上一章within配置的时间段），默认来说会被丢弃。如果我们需要捕获这种超时的匹配结果，可以使用自定义的PatternProcessFunction，实现TimedOutPartialMatchHandler。如下所示：
    

    class CustomPatternProcessFunction extends PatternProcessFunction<Object, Object> implements TimedOutPartialMatchHandler<Object> {
    @Override
    public void processMatch(Map<String, List<Object>> map, Context context, Collector<Object> collector) throws Exception {
        // ...
    @Override
    public void processTimedOutMatch(Map<String, List<Object>> map, Context context) throws Exception {
        Object element = map.get("key").get(0);
        context.output(outputTag, element);

    上面的例子将超时的element放入旁路数据，绑定到一个outputTag上。OutputTag用于标记一组旁路输出的元素。下面是创建OutputTag和获取旁路数据元素的方法：
    

    // Object为旁路输出的元素类型
OutputTag<Object> outputTag = new OutputTag<Object>("late-element"){};
// CEP操作...
DataStream<Object> sideOutput = patternStream.getSideOutput(outputTag);

    如果使用event time模式，一定会有来迟的元素。如果我们需要对这些元素进行捕获处理，可以和上面一样，使用旁路输出：
    

    patternStream.sideOutputLateData(lateDataOutputTag)