2021-07-29

《Flink 实战与性能优化》—— 使用 Flink CEP 处理复杂事件

6.2 使用 Flink CEP 处理复杂事件

6.1 节中介绍 Flink CEP 和其使用场景，本节将详细介绍 Flink CEP 的 API，教会大家如何去使用 Flink CEP。

6.2.1 准备依赖

要开发 Flink CEP 应用程序，首先你得在项目的 pom.xml 中添加依赖。

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

这个依赖有两种，一个是 Java 版本的，一个是 Scala 版本，你可以根据项目的开发语言自行选择。

6.2.2 Flink CEP 入门应用程序

准备好依赖后，我们开始第一个 Flink CEP 应用程序，这里我们只做一个简单的数据流匹配，当匹配成功后将匹配的两条数据打印出来。首先定义实体类 Event 如下：

public class Event {
    private Integer id;
    private String name;
}

然后构造读取 Socket 数据流将数据进行转换成 Event，代码如下：

SingleOutputStreamOperator<Event> eventDataStream = env.socketTextStream("127.0.0.1", 9200)
    .flatMap(new FlatMapFunction<String, Event>() {
        @Override
        public void flatMap(String s, Collector<Event> collector) throws Exception {
            if (StringUtil.isNotEmpty(s)) {
                String[] split = s.split(",");
                if (split.length == 2) {
                    collector.collect(new Event(Integer.valueOf(split[0]), split[1]));
                }
            }
        }
    });

接着就是定义 CEP 中的匹配规则了，下面的规则表示第一个事件的 id 为 42，紧接着的第二个事件 id 要大于 10，满足这样的连续两个事件才会将这两条数据进行打印出来。

Pattern<Event, ?> pattern = Pattern.<Event>begin("start").where(
        new SimpleCondition<Event>() {
            @Override
            public boolean filter(Event event) {
                log.info("start {}", event.getId());
                return event.getId() == 42;
            }
        }
).next("middle").where(
        new SimpleCondition<Event>() {
            @Override
            public boolean filter(Event event) {
                log.info("middle {}", event.getId());
                return event.getId() >= 10;
            }
        }
);

CEP.pattern(eventDataStream, pattern).select(new PatternSelectFunction<Event, String>() {
    @Override
    public String select(Map<String, List<Event>> p) throws Exception {
        StringBuilder builder = new StringBuilder();
        log.info("p = {}", p);
        builder.append(p.get("start").get(0).getId()).append(",").append(p.get("start").get(0).getName()).append("\n")
                .append(p.get("middle").get(0).getId()).append(",").append(p.get("middle").get(0).getName());
        return builder.toString();
    }
}).print();//打印结果

然后笔者在终端开启 Socket，输入的两条数据如下：

1 2	42,zhisheng 20,zhisheng

作业打印出来的日志如下图所示：

整个作业 print 出来的结果如下图所示：

好了，一个完整的 Flink CEP 应用程序如上，相信你也能大概理解上面的代码，接着来详细的讲解一下 Flink CEP 中的 Pattern API。

6.2.3 Pattern API

你可以通过 Pattern API 去定义从流数据中匹配事件的 Pattern，每个复杂 Pattern 都是由多个简单的 Pattern 组成的，拿前面入门的应用来讲，它就是由 start 和 middle 两个简单的 Pattern 组成的，在其每个 Pattern 中都只是简单的处理了流数据。在处理的过程中需要标示该 Pattern 的名称，以便后续可以使用该名称来获取匹配到的数据，如 p.get("start").get(0) 它就可以获取到 Pattern 中匹配的第一个事件。接下来我们先来看下简单的 Pattern 。

单个 Pattern

数量

条件

组合 Pattern

Group Pattern

事件匹配跳过策略

6.2.4 检测 Pattern

选择 Pattern

6.2.5 CEP 时间属性

根据事件时间处理延迟数据

时间上下文

加入知识星球可以看到上面文章：https://t.zsxq.com/nMR7ufq

6.2.6 小结与反思

本节开始通过一个 Flink CEP 案例教大家上手，后面通过讲解 Flink CEP 的 Pattern API，更多详细的还是得去看官网文档，其实也建议大家好好的跟着官网的文档过一遍所有的 API，并跟着敲一些样例来实现，这样在开发需求的时候才能够及时的想到什么场景下该使用哪种 API，接着教了大家如何将 Pattern 与数据流结合起来匹配并获取匹配的数据，最后讲了下 CEP 中的时间概念。

你公司有使用 Flink CEP 吗？通常使用哪些 API 居多？

赏

纯属好玩