128.尚硅谷_Flink项目-电商用户行为分析_CEP简介（四）_模式的检测和事件处理

那我们已经知道详细了解了个体模式和模式序列，那我们再来说一下在代码里边到底是做了什么操作，那在代码里边其实主要的操作呢，首先我们要定义一个模式pattern，对吧？前面我们已经讲了怎么定义个体模式，怎么定义模式序列，这个定义好之后，下一步操作其实就是要做模式的检测，或者说要把这个模式要应用到当前的我们的那个输入的事件流上，Data stream上啊，那这里边一般我们应用的是什么呢？你可以直接应用到data stream上，也可以应用到kid stream上，我们一般情况还是要做皮的啊，还是要先分组，因为你接下来做的这个事件模式的检测，一般情况你像我们这里边都是针对同一个user ID，对吧？你不可能把不同的user混在一起去做这个事件的匹配嘛，啊，所以往往我们的场景也都是做这个K之后再去做操作的，然后接下来我们这里边，呃，就是做这个模式。

检测的时候啊，那主要就是CP调CP.python方法得到一个Python stream，然后里边我们其实非常简单，就是把当前的data stream input的数据流和当前定义好的pattern作为参数传进去就完事了啊，应用在当前这个数据流上对吧？啊，就相当于现在我们就可以直接摘出，还记得之前那个图吧，直接可以摘出来想要的这个数据，但是现在呢，还没有生成真正的就是我们最后得想得到的输出应该是一个data stream，对吧？大家还记得这个图最后想得到的这样一个data stream，那现在呢，我还没得到这样一个data stream，只是相当于框了一下，把它框起来了，对吧，就这张图上，在这里我们把它框起来了，已经检测到了，然后接下来怎么样把它摘出来，然后再我们还想做一些转换，包装成想要的样例类，对吧，因为之前给大家说过，存的时候是用一个map类型，数据类型来存着的嘛，所以现在我们还想做一些转换操作那下。

它一步检测到了之后的下一步啊，那就是要做一个提取了，匹配事件的提取，那这个就可以应用一个select或者Fla select的方法，那另外还有一个特殊的叫process啊，大家知道process就是底层的process API，这个process function嘛，啊，跟我们之前讲的这个底层A是类似的啊，啊基本上大家可以认为就是一个特殊的select或者Fla select，只不过里边可以应用到这个，呃，调用到上下文里边的一些东西，对吧，可以有更多的这个操作，那我们一般使用呢，就用这个select select select select就可以，那最常规的使用方式select呢，可以传一个参数，就是pattern select function，对吧，大家还记得就是传这么一个东西啊，然后这个pattern select function，它的含义是什么呢？啊，就是具体检测到的对应的那个事件序列，接下来要做的处理操作，也就是说什么呢？每个成功匹配到的事件序列都会调用到接。

它来我们定义的这个方式对吧？啊，所以大家看到这里边输入的参数就是一个map嘛，这个map类型，我们说它的key是string string是什么呢？就是我们定义模式序列里边每一个个体模式的那个名称对吧？那个字符串，然后里边接接收到的这个数据value啊，这个数据类型又是什么呢？是一个interable类型，大家看到底层其实就是一个抓list对吧？嗯，那么是一个list子的类型，为什么是一个list子类型呢？因为我们当时检测到的个体模式对应的事件有可能是一个，有可能是多个，这要看你到底有没有定义那个循环模式对吧？如果说你定义了循环模式，检测到的就是多个，放到这个例子里边就可以了，好啊，那所以里边我们做的转换就可以从这个map里边把想要的事件都拿出来，然后包装成自己想要的这个输出的数据，结果输出就完事了啊，这个就是非常简单的一个操作啊，那如果要是回到之前这幅图里边来的话。

大家其实就想到了，那就是把这个事件这个就是我们的那一个map对不对啊，检测到这个事件，大家想到本身这也是流嘛，那我是不是应该也是检测到一个，在后边这个流里边就会输出一个啊，就是在我们得到这个data stream里边检测到一个，就应该输出一个，对吧？输出一个，诶，这是一个map数据类型，是不是就会调用一次我们定义的那个pattern select方式，哎，就会调一个那个方式对吧？直接把它做一个转换得到的，大家会想到得到的是不是就是自己定义好的另外的一个，比方说我，我管这个变成这样的一个数据类型，对吧，我map成这样的一个数据类型，做一个输出啊，这就是我们最后包装成的一个样例类嘛，整个的处理流程就是这样的啊，所以大家可以再稍微的总结一下之前我们的代码里面做的那些操作，然后接下来还要给大家说另外一个比较特殊的场景，就是关于这个时间的处理啊，前面我们不是说过有那个within吗？

Within的话，相当于是在这个时间范围内，我们要把前面定义好的那个序列都匹配上，对吧？那就涉及到一个问题，就是假如说在这个时间范围内每一匹配上呢？没匹配上就有一个问题，大家知道那个时间相当于是定注册了一个定时器一样嘛，那如果或者说大家可以认为是相当于对于当前开了一个窗口一样啊，那如果没匹配上的话，那其实我们知道对于这个定义的Python啊模式而言，其实前面有可能是什么呢？哎，前面的几个，比方说我那个啊，Start啊，第一个第一步是什么情况，然后后面接下来next做什么操作啊，Middle中间做什么操作也可配上了，只不过就等最后一步那个and的时候没等来，然后你就到点了，对吧？哎，微定十秒钟，十秒钟到时了，哎，那这种情况就会出现什么呢？就是开头正常还是匹配上的，本来我还可以继续等，但是超时然后没匹配上对吧？哎，在有些场景下，这其实是值得考虑的一个问题。

呃，就是说呃，我我有时候需要去知道，就是哪些是真的没匹配上的，哪些是匹配上的，而哪些呢是开始匹配上了，但是超时了，对吧，最后没有匹配成功的啊，所以对于这个超时事件，如果有时候我们关心的话，也可以把它单独的提取出来，那这种提取的方式是什么呢？哎，那那那就是说呃，大家可以认为啊，就是说在这个呃，Cp patternon API里边给我们又提供了一个测殊出流，就把超时的事件会直接放到一个特殊出流里面去，所以接下来我们提取的方式呢，你还是要定义一个测输出流的标签，然后接下来大家看到这个提取的过程就不太一样了，调用的这个方法之前我们看也看到啊，Select的方法里边有各种不同的传参方式啊，可以做这个重载啊，那这里边有一种方式是什么呢？大家看这是这个。

笔画的这种传餐方式啊，前面给一个output taag对吧，Select直接给一个out ta，然后后边呢，接着再传两个方式，哎，这两个function式分别代表什么呢？前面这个function式大家看到有一个time stamp对吧？比后面这个function不太一样了，哎，这个有一个time stamp，这里代表什么含义呢？代表就是当前触发的那个时间，哎，所以当前的这个有时间戳嘛，呃，有那个出发的时间，所以它要处理的是超时的时间，也就是说当前的这个这个function啊，它是一个pattern timeout function式，专门用于处理超时事件的方式，然后后边还有一个方式，这个大家就比较熟悉了，一个输入参数就是一个map对吧，然后输出我们最后匹配上的一个一个结果啊，包装好的一个数据类型，这就是我们要传的一个pattern select方式，就用这种方式去把它区分开。

啊，所以大家可能知道它具体做的这个操作，对应到前面我们那张图里边的话，诶，那就是这里边我们先检测对吧？呃，应用这个模式之后，我们先检测到，然后接下来呢，如果说如果说啊，这里边这个检测到的成功匹配到的事件，我就直接做这个转换之后输出到后边的这个主流data stream里边来，那假如说这个超时的事件怎么办呢？诶超时的事件一开始我这个其实也都在缓存的，是缓存着呢，对吧？哎，但是到那个点的时候，我发现超时了，不等了，不等了呢，那些缓存的事件呢，不要丢，把它怎么样分到一个测输出流里边来，这个测输出流里边，哎，可能就是就是一部分数据对吧？哎，大家就想到这个，我给半个这个三角啊，给一个直角三角形表示的是什么呢？可能就只检测到我们前面的一部分匹配上的是实间对吧，但它也不是不匹配，为什么呢？

因为后边那个数只是没来，只是现在你那个规定好的那个时间到了而已嘛，我就直接把它输出到这个测试出流了啊这一部分我们接下来怎么去做操作呢？大家看到这里的数据，我们做的操作是用pattern select function对吧？哎，然后这里边的这个操作我们用的是pattern t out方式对吧？定义两种不同的操作，各自处理各自就完事了，然后最最后怎么样再提取呢？那当然就是定义了那个auto tag嘛，测输出流的标签，那最后我们就是单独再去get set out food，用测出测输出流标签，把对应的那个流里边的数据提取出来就完事了啊，这就是这个完整的处理流程啊，CP调用这个判断API进行复杂事件处理的过程。