Flink_CEP_4步骤

末蓝、 2022-11-22 00:12 185阅读 0赞

*  1.首先定义一个匹配的序列模式
 *  2.将pattern应用在数据流上
 *  3.定义一个测输出流标签，用来标明超时事件的测输出流
 *  4.调用select方法，提取匹配事件和超时事件，分别进行处理转换输出  
    自定义超时时间继承PatternTimeoutFunction  
    自定义匹配处理函数PatternSelectFunction

//1.定义一个匹配事件序列的模式
        val orderPayPattern = Pattern
          .begin[OrderEvent]("create").where(_.eventType == "create")  //首先是订单的create事件
          .followedBy("pay").where(_.eventType == "pay") //后面来的是订单的pay事件
          .within(Time.minutes(15))
    
        //2.将pattern应用在按照orderId分组的数据流上
        val patterStream = CEP.pattern(orderEventStream.keyBy(_.orderId), orderPayPattern)
    
        //3.定义一个测输出流标签，用来标明超时事件的测输出流
        val orderTimeOutputTag = new OutputTag[OrderResult]("order time out")
    
        //4.调用select方法，提取匹配事件和超时事件，分别进行处理转换输出
        val result = patterStream
          .select(orderTimeOutputTag, new OrderTimeOutSelect(), new OrderPaySelect())

如果没看懂，提供一个案列  
[https://blog.csdn.net/qq\_46548855/article/details/109462736][https_blog.csdn.net_qq_46548855_article_details_109462736]  
Flink官网CEP  
[https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/libs/cep.html][https_ci.apache.org_projects_flink_flink-docs-release-1.11_dev_libs_cep.html]

[https_blog.csdn.net_qq_46548855_article_details_109462736]: https://blog.csdn.net/qq_46548855/article/details/109462736
[https_ci.apache.org_projects_flink_flink-docs-release-1.11_dev_libs_cep.html]: https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/libs/cep.html