阅读 1565

探索如何使用Flink CEP

写在前面

前言的前言

在学习Flink的过程中,我看过很多教程。无论是视频还是博文,几乎都把Flink CEP作为进阶内容来讲授。究其原因,大概是CEP涉及到的计算机基础知识很多,而我对于诸如NFA、DFA之类名词的印象,基本只停留在很多年前编译原理的课本上。那么如何在仅了解有限的基础知识的情况下,快速使用Flink CEP完成开发呢?

实际上,在我的学习过程中,最大的困难来自于对匹配规则的理解。在此基础上,我进行了大量实验,再结合部分代码的阅读,最终窥见冰山一角。需要说明的是,本文的所有实验,均建立在默认跳过策略,也就是AfterMatchSkipStrategy为noSkip的情况。 至于其他跳过策略下匹配方式的区别,也许有必要单独写一篇文章来说明。

准备工作

为了简化流程,方便查看运行过程和结果,我手动实现了一个SourceFunction。该SourceFunction每次读入一行,随后休眠100毫秒(为方便看到输出),代码如下:

public class ReadLineSource implements SourceFunction<String> {
    private String filePath;
    private boolean canceled = false;

    public ReadLineSource(String filePath) {
        this.filePath = filePath;
    }

    @Override
    public void run(SourceContext<String> sourceContext) throws Exception {
        var reader = new BufferedReader(new FileReader(filePath));
        while (!canceled && reader.ready()) {
            var line = reader.readLine().strip();
            System.out.println("读取:".concat(line));
            sourceContext.collect(line);
            Thread.sleep(100);
        }
        sourceContext.close();
    }

    @Override
    public void cancel() {
        canceled = true;
    }
}
复制代码

同时,引入Maven依赖以便开发。本文使用Java语言。

<properties>
    <flink.version>1.9.1</flink.version>
</properties>
<dependencies>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-java</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-java_2.12</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-cep_2.12</artifactId>
        <version>${flink.version}</version>
    </dependency>
</dependencies>
复制代码

从单个Pattern入手

我们先从简单的内容入手。看看在单个Pattern下,Flink CEP是如何匹配的。

各个API的用法

在学习Flink CEP的过程中,很容易找到相似的博文,文章中使用表格列举出了各个API的作用。然而大家很容易发现,这东西太像正则表达式了(实际上底层匹配逻辑的实现方式应该也和正则表达式类似)。因此,结合正则表达式理解这些API显得十分快速,所以我自作主张,加上了功能相近的正则表达式。例如,我们要用CEP匹配字母x:

API 含义 正则表达式 示例
where() 指定匹配条件 x pattern.where(_ = "x"), 匹配x
times() 模式发生次数 {2,4} pattern.times(2,4), 发生2,3,4次
timesOrMore() 模式发生次数>=x {2,} pattern.timesOrMore(2), 发生2次及以上
oneOrMore() 模式发生一次及以上 + pattern.oneOrMore(), 发生一次及以上
optional() 模式可以不匹配 {2}? pattern.times(2).optional(), 发生0或2次
greedy() 贪心匹配 {2,} pattern.times(2).greedy()
or() 模式的或条件 (x|y) pattern.where(_ = "y"), 匹配y
until() 停止条件 x+?y pattern.oneOrMore().until(_ = "y")

仅使用where和or写一个程序

下面我们使用where和or,写出第一个匹配程序。该程序匹配输入流中,所有以x或y开头的数据。

public class CepDemo {
    public static void main(String[] args) throws Exception {
        var environment = StreamExecutionEnvironment.getExecutionEnvironment();
        var stream = environment.setParallelism(1).addSource(new ReadLineSource("Data.txt"));

        var pattern = Pattern.<String>begin("start").where(new IterativeCondition<>() {
            @Override
            public boolean filter(String s, Context<String> context) {
                return s.startsWith("x");
            }
        }).or(new IterativeCondition<>() {
            @Override
            public boolean filter(String s, Context<String> context) throws Exception {
                return s.startsWith("y");
            }
        });

        CEP.pattern(stream, pattern).select((map ->
                Arrays.toString(map.get("start").toArray()))
        ).addSink(new SinkFunction<>() {
            @Override
            public void invoke(String value, Context context) {
                System.out.println(value);
            }
        });
        environment.execute();
    }
}
复制代码

随后,在Data.txt中写入如下内容:

x1
z2
c3
y4

运行程序,输出如下结果:

读取:x1   
[x1]   
读取:z2   
读取:c3   
读取:y4   
[y4]
复制代码

此时可以发现,Flink CEP是根据输入的每一条数据进行匹配的。单条数据可以是本文中的字符串,也可以是复杂的事件对象,当然也可以是字符。如果每一条数据都是一个字符,那CEP就和正则表达式十分相似了。所以我姑且将它当作正则表达式Pro Plus。

加上量词

接下来,还是在单个Pattern中,我们加上量词API,研究研究Flink CEP是如何匹配多条数据的。从这里开始,事情和正则表达式有了一些差距。差距主要在结果的数量上。由于是流计算,因此在实际处理过程中,Flink无法知道后续的数据,所以会输出所有匹配的结果。

例如,使用timesOrMore()函数,匹配以a开头的字符串出现3次及以上的情况,首先编写代码(其他代码与上方的例子完全一致,为节约篇幅不再列出,下同):

var pattern = Pattern.<String>begin("start").where(new IterativeCondition<>() {
    @Override
    public boolean filter(String s, Context<String> context) {
        return s.startsWith("a");
    }
}).timesOrMore(3);
复制代码

随后在Data.txt中输入如下字符串序列:

a1
a2
a3
b1
a4

运行程序,输出如下结果:

读取:a1
读取:a2
读取:a3
[a1, a2, a3]
读取:b1
读取:a4
[a1, a2, a3, a4]
[a2, a3, a4]
复制代码

下面分析一下执行流程。程序开始后,等待数据流入。当a1和a2输入后,由于暂时不满足条件,所以没有产生结果,只是将数据储存在状态中。a3到来后,第一次满足了匹配条件,因此程序输出结果 [a1, a2, a3]。随后,b1输入,不满足条件;接下来a4输入。此时,a1、a2和a3依旧储存在状态中,因此依然可以参与匹配。匹配可以产生多个结果,但是有两个原则:

  1. 必须严格按照数据流入的顺序;
  2. 产生的结果必须包含当前元素;

原则1很好理解,由于数据的流入是按照a1 -> a2 -> a3 -> a4的顺序,所以结果生成的序列也必须按照这个顺序,不能删减中间数据,更不能打乱顺序。因此, [a1, a2, a4][a3, a2, a4, a1] 这种结果是不可能生成的。原则2就更好理解了,数据是因为a4的流入才产生的,再考虑到我们设定的量词条件是“三个及以上”,因此产生的结果只可能是 [a2, a3, a4][a1, a2, a3, a4]

同理,如果我们在Data.txt最后加入一行a5,则程序输出结果如下:

读取:a1
读取:a2
读取:a3
[a1, a2, a3]
读取:b1
读取:a4
[a1, a2, a3, a4]
[a2, a3, a4]
读取:a5
[a1, a2, a3, a4, a5]
[a2, a3, a4, a5]
[a3, a4, a5]
复制代码

按照这种思路,如果我们继续加上a6、a7、a8、……、a100,那么每个数据产生的结果会越来越多,因为Flink CEP会把所有符合条件的数据储存在状态里。这样下去不行的,要不然内存养不起它的(By 华农兄弟)。因此,oneOrMore()和timesOrMore()之类的函数后面,一般都要跟上until()函数,从而指定终止条件。

把量词换成times()

如果使用和上面一样的数据,但是把量词换成times(3),会产生什么样的结果?我们首先修改代码:

var pattern = Pattern.<String>begin("start").where(new IterativeCondition<>() {
    @Override
    public boolean filter(String s, Context<String> context) {
        return s.startsWith("a");
    }
}).times(3);
复制代码

由于固定了只匹配三个,再加上前文提到的两个原则的束缚,结果就很明显了:

读取:a1
读取:a2
读取:a3
[a1, a2, a3]
读取:b1
读取:a4
[a2, a3, a4]
读取:a5
[a3, a4, a5]
复制代码

从a1到b1的逻辑完全相同,当读取到a4时,由于只匹配3个,同时结果必须包含a4,因此产生的结果只能是 [a2, a3, a4] 。同理读取到a5后,由于结果必须包含a5且只匹配3个,所以结果只能是 [a3, a4, a5] 。这种情况下,过期的数据会被清理掉,妈妈再也不用担心我的内存不够用了。

除了固定参数,times()函数还支持times(from, to)指定边界。这种情况下的匹配结果和上文类似,相信大家很容易就能推出来,在此我就不再赘述了。

使用严格模式

大家也许注意到,上文的Data.txt中,一直有一个讨厌的b1。由于不满足我们的基本匹配条件,b1直接被我们的程序忽略掉了。这是因为Flink CEP默认采用了不严格的匹配模式,而在某些情况下,这种数据是不能忽略的,这时候就可以使用consecutive()函数,指定严格的匹配模式。修改代码如下:

var pattern = Pattern.<String>begin("start").where(new IterativeCondition<>() {
    @Override
    public boolean filter(String s, Context<String> context) {
        return s.startsWith("a");
    }
}).times(3).consecutive();
复制代码

运行程序,产生如下结果:

读取:a1
读取:a2
读取:a3
[a1, a2, a3]
读取:b1
读取:a4
读取:a5
复制代码

此时,由于a1、a2、a3是紧密相连的,因此被成功匹配。而a2、a3、a4和a3、a4、a5中间由于多了一个b1,在严格模式下不能被匹配。可以看出,严格模式下的匹配策略更像正则表达式。

多来几个Pattern

一般而言,需要使用CEP的任务都得依靠多个Pattern才能解决。此时,可以使用followedBy()、next()等函数创建一个新的Pattern,并按照不同的逻辑将新Pattern和前一个Pattern连接起来。

使用followedBy()创建一个新的Pattern

在上文的基础上,我们修改匹配逻辑,使得新逻辑能匹配“包含2-3个a开头的字符串,同时包含1-2个b开头的字符串”的输入数据。修改代码如下:

var pattern = Pattern.<String>begin("start").where(new IterativeCondition<>() {
    @Override
    public boolean filter(String s, Context<String> context) {
        return s.startsWith("a");
    }
}).times(2, 3).followedBy("middle").where(new IterativeCondition<String>() {
    @Override
    public boolean filter(String s, Context<String> context) throws Exception {
        return s.startsWith("b");
    }
}).times(1, 2);

CEP.pattern(stream, pattern).select(map -> {
    var list = map.get("start");
    list.addAll(map.get("middle"));
    return Arrays.toString(list.toArray());
}).addSink(new SinkFunction<>() {
    @Override
    public void invoke(String value, Context context) {
        System.out.println(value);
    }
});
复制代码

这里我们使用了followedBy()函数,该函数创建了一个名为“middle”的新Pattern,新Pattern中包含了指向原Pattern的引用。同样发生变化的是select函数中的lambda表达式。在表达式中,我们除了获取名为“start”的Pattern中的数据,还获取了名为“middle”的Pattern的数据,并将他们拼在一起。这与正则表达式中的子表达式特别类似,实际上,我们可以将每个Pattern近似看作一个子表达式,在读取结果的时候,使用Pattern的名字,从map中提取出结果。

在修改过代码后,我们修改Data.txt中数据如下:

a1
a2
a3
b1
a4
a5
b2

运行程序,得到如下结果:

读取:a1
读取:a2
读取:a3
读取:b1
[a1, a2, a3, b1]
[a1, a2, b1]
[a2, a3, b1]
读取:a4
读取:a5
读取:b2
[a1, a2, a3, b1, b2]
[a1, a2, b1, b2]
[a2, a3, a4, b2]
[a2, a3, b1, b2]
[a3, a4, a5, b2]
[a3, a4, b2]
[a4, a5, b2]
复制代码

一下子产生了这么多数据,我一开始还是很懵的。接下来我们逐步分析下:

  1. a1, a2依次读入,不满足整体条件,但是满足“start”条件,且产生了 [a1, a2] 这一中间结果,存在状态中;
  2. a3读入,不满足整体条件,但是满足“start”条件,且产生了 [a2, a3][a1, a2, a3] 两个结果;
  3. b1读入,满足“middle”条件,产生 [b1] 中间结果。此时整体条件满足,因此和上述中间结果组合输出 [a1, a2, a3, b1][a1, a2, b1][a2, a3, b1]
  4. a4读入,继续满足“start”条件,产生 [a2, a3, a4][a3, a4]; 两个结果,但是由于这两个结果是在b1读入之后产生的,因此这两个结果不能和 [b1] 进行组合;
  5. a5读入,继续满足“start”条件,产生 [a3, a4, a5][a4, a5] 两个中间结果,同理不能和 [b1] 进行组合;
  6. b2读入,继续满足“middle”条件,产生 [b1, b2][b2] 两个中间结果。这里开始比较复杂了,需要严格结合时间顺序来分析。由于b1是在a4之前读入的,因此包含b1的序列 [b1, b2] 只能与 [a1, a2][a2, a3][a1, a2, a3] 进行关联。而 [b2] 则可以与包含了a4或a5的 [a2, a3, a4][a3, a4][a3, a4, a5][a4, a5] 四个序列关联,因此此时输出结果如下:
[a1, a2, a3, b1, b2]    // [a1, a2, a3] 和 [b1, b2] 关联
[a1, a2, b1, b2]        // [a1, a2] 和 [b1, b2] 关联
[a2, a3, a4, b2]        // [a2, a3, a4] 和 [b2] 关联
[a2, a3, b1, b2]        // [a2, a3] 和 [b1, b2] 关联
[a3, a4, a5, b2]        // [a3, a4, a5] 和 [b2] 关联
[a3, a4, b2]            // [a3, a4] 和 [b2] 关联
[a4, a5, b2]            // [a4, a5] 和 [b2] 关联
复制代码

那么有一个问题,为什么 [b2] 不能与 [a1, a2][a2, a3][a1, a2, a3] 进行关联呢?还是要站在时间序列的角度进行解释。因为只有b1是跟随在这三个元素后面的,所以只有包含b1的两个序列([b1][b1, b2])可以和它们进行关联,这就是followedBy的含义。为了验证这一观点,我们在Data.txt最后加上一个b3,在其他代码均不变的情况下,最后读入b3后,输出如下结果:

[a2, a3, a4, b2, b3]
[a3, a4, a5, b2, b3]
[a3, a4, b2, b3]
[a4, a5, b2, b3]
复制代码

分析如下:当读入b3后,满足“middle”条件,生成 [b2, b3][b3]。其中,只有 [b2, b3] 包含了b2,由于b2是距离 [a2, a3, a4][a3, a4][a3, a4, a5][a4, a5] 四个序列最近的数据,因此只有 [b2, b3] 才能和上述四个序列关联。而 [b3] 由于不包含b2,因此无法和它们关联。

将followedBy()换成next()

可以将next()看作是加强版的followedBy()。在followedBy中,两个Pattern直接允许不紧密连接,例如上文中的 [a1, a2][b1] ,他们中间隔了一个a3.这种数据在next()中会被丢弃掉。使用上文同样的数据(不包括b3),将代码中的followedBy换成next,修改如下:

var pattern = Pattern.<String>begin("start").where(new IterativeCondition<>() {
    @Override
    public boolean filter(String s, Context<String> context) {
        return s.startsWith("a");
    }
}).times(2, 3).next("middle").where(new IterativeCondition<String>() {
    @Override
    public boolean filter(String s, Context<String> context) throws Exception {
        return s.startsWith("b");
    }
}).times(1, 2);
复制代码

运行后,看到如下结果:

读取:a1
读取:a2
读取:a3
读取:b1
[a1, a2, a3, b1]
[a2, a3, b1]
读取:a4
读取:a5
读取:b2
[a1, a2, a3, b1, b2]
[a2, a3, b1, b2]
[a3, a4, a5, b2]
[a4, a5, b2]
复制代码

和之前的结果进行分析,发现结果中的 [a1, a2, b1][a1, a2, b1, b2][a2, a3, a4, b2][a3, a4, b2] 均被排除,因为他们相比原序列,分别缺少了a3、a3、a5、a5。

greedy()做了什么

关于greedy()的用法,可以说是十分令人迷惑的。我看了许多文章,对greedy()的描述几乎都是一笔带过。描述大多是“尽可能多的匹配”,但是实际上,大多数情况下加不加greedy()几乎没有任何区别。因为greedy()虽然被归为量词API,但是它实际上是在多个Pattern中才能起作用的。 为此,我找到了greedy()的实现逻辑,在NFACompiler类的updateWithGreedyCondition方法中,代码如下:

private void updateWithGreedyCondition(
	State<T> state,
	IterativeCondition<T> takeCondition) {
	for (StateTransition<T> stateTransition : state.getStateTransitions()) {
		stateTransition.setCondition(
			new RichAndCondition<>(stateTransition.getCondition(), 
			new RichNotCondition<>(takeCondition)));
	}
}
复制代码

阅读代码,发现该方法实际上添加了一个逻辑:确认当前条件满足转换到下一个state所需的条件,且不满足当前state的条件。意思就是,如果当前处于Pattern1,但是出现了一条同时满足两个Pattern1和Pattern2条件的数据,在不加greedy()的情况下,会跳转到Pattern2,但是如果加了greedy(),则会留在Pattern1。下面我们来验证一下,编写如下代码:

var pattern = Pattern.<String>begin("start").where(new IterativeCondition<>() {
    @Override
    public boolean filter(String s, Context<String> context) {
        return s.startsWith("a");
    }
}).times(2, 3).next("middle").where(new IterativeCondition<String>() {
    @Override
    public boolean filter(String s, Context<String> context) throws Exception {
        return s.length() == 3;
    }
}).times(1, 2);
复制代码

在这一代码中,如果一条数据为a开头,且长度为3,则同时满足“start”和“middle”。同时,为了方便区分数据到底属于哪个Pattern,我们在输出前加入分隔符:

CEP.pattern(stream, pattern).select(map -> {
    var list = map.get("start");
    list.add("|");
    list.addAll(map.get("middle"));
    return Arrays.toString(list.toArray());
}).addSink(new SinkFunction<>() {
    @Override
    public void invoke(String value, Context context) {
        System.out.println(value);
    }
});
复制代码

准备如下数据:

a
a1
a22
b33

在不加greedy()的情况下,运行结果如下:

读取:a
读取:a1
读取:a22
[a, a1, |, a22]
读取:b33
[a, a1, a22, |, b33]
[a, a1, |, a22, b33]
[a1, a22, |, b33]
复制代码

观察结果,可知a22在两个Pattern中左右横跳,输出了所有可能的结果。接下来我们加上greedy():

var pattern = Pattern.<String>begin("start").where(new IterativeCondition<>() {
    @Override
    public boolean filter(String s, Context<String> context) {
        return s.startsWith("a");
    }
}).times(2, 3).greedy().next("middle").where(new IterativeCondition<String>() {
    @Override
    public boolean filter(String s, Context<String> context) throws Exception {
        return s.length() == 3;
    }
}).times(1, 2);
复制代码

运行结果如下:

读取:a
读取:a1
读取:a22
读取:b33
[a, a1, a22, |, b33]
[a1, a22, |, b33]
复制代码

此时,a22被划到了“start”这一Pattern中。由此可见,greedy()影响的是“同时满足两个Pattern条件的数据的划分逻辑”,而且加了greedy()后,产生的结果会变少,并不是直观印象中的,产生尽可能多条的数据。

总结

在我看来,Flink CEP的难度主要集中在对执行过程和多个Pattern之间匹配流程的理解。虽然在很多的例子中,即使对匹配流程不求甚解也能得到正确的结果,但是掌握之后必然能够提高开发效率,以及程序的可靠性。

本文只介绍了少数几个API,权当抛砖引玉。不过我相信,通过这些API,能够很好的揭开Flink CEP的面纱,至于其他API的用法,在文档的帮助下很容易就能理解。同时,由于时间和我的水平所限,对于部分内容的理解很可能存在偏差或者错误,希望大神看到后能够及时指出,以帮助我改正错误。

本文所有内容均基于默认的跳过策略,也许当我弄明白不同跳过策略的具体差异后,能再水一篇文章出来……最后,感谢大家阅读。这是我在掘金上写的第一篇文章,也是我写的第一篇Flink系列的文章。希望在后续的学习和工作中,能够进一步加深对Flink的理解,从而总结出更多的文章来与大家交流。