Flink人生几何？Flink学习指南

Flink 使用之 CEP

2021-09-23  本文已影响0人  AlienPaul

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个方法：

指定重复次数

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

配置的方法为：

指定条件

匹配后跳过策略

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

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

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

Pattern.begin("patternName", skipStrategy);

skipStrategy通过如下方式创建：

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

时间限制

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

创建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<>("late-element");

// CEP操作...

DataStream<Object> sideOutput = patternStream.getSideOutput(outputTag);

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

patternStream.sideOutputLateData(lateDataOutputTag)

新版变化

从Flink 1.12开始，CEP默认从Processing Time改为Event Time。使用时务必要注意。详情参见Flink 升级1.12版本的坑。

使用SQL方式编写CEP

除了使用Pattern API，Flink还支持使用SQL方式编写CEP，相比而言SQL更为灵活，但是需要学习SQL match_recognize子句的语法。SQL方式编写CEP参见Flink 使用之 CEP（SQL方式）。

上一篇 下一篇

猜你喜欢

热点阅读