一、简介

flink CEP是flink中的一个事件处理模块，可以自动在流式数据处理中定义并自动检索事件，从其他帖子中了解到滴滴打车用户行为就是用了flink的这个模块，举个简单的例子，假设需要一分钟内一个账号连续登录三次密码错误既冻结该账户，当成千上万的用户不断地登录时可以把所有的登录请求数据视为一个流式数据，把用户名或用户id设为key，flink就可以自动检索“一分钟内一个账号连续三次登录密码错误”这件事并给出对应的三次信息供后续操作，flink在整体设计上比较像函数式编程，flink CEP在设计上也是如此，因此对整个代码的侵入性非常小，这里的demo会在上一篇（https://blog.csdn.net/qq_37720936/article/details/107044407）kafka到数据库的demo上进行拓展。

二、添加依赖

由于创建的模板工程中本身不带CEP模块，需要添加CEP模块的依赖

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

三、代码示例

本例中事件模板为当五秒中出现三次名字为z3的数据之后，接下来再出现两次名字为z3的数据，则触发该事件，先上代码：

public static void main(String[] args) throws Exception { // set up the streaming execution environment final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); Properties properties = new Properties(); properties.setProperty("bootstrap.servers", "localhost:9092"); FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>("test", new SimpleStringSchema(), properties); // 设置消费模式 consumer.setStartFromGroupOffsets(); DataStream<String> stream = env .addSource(consumer) .filter(s -> { try { JSONObject.parseObject(s); } catch (JSONException e) { return false; } return !StrUtil.isBlank(s); }); stream.addSink(new RichSinkFunction<String>() { private Connection connection; @Override public void open(Configuration parameters) throws Exception { connection = DriverManager.getConnection("jdbc:sqlserver://ip", "userName", "password"); } @Override public void invoke(String value, Context context) throws Exception { JSONObject jsonObject = JSON.parseObject(value); connection.createStatement().execute("INSERT gmy.dbo.flinkTest (id, name, age, time) VALUES ('" + jsonObject.getString("id") + "','" + jsonObject.getString("name") + "','" + jsonObject.getString("age") + "','" + new Date(jsonObject.getLong("time")) + "')"); } @Override public void close() throws SQLException { connection.close(); } }); PatternStream<String> nameTimesPatternStream = getNameTimesPatternStream(stream, getNameTimesPattern()); DataStream<String> eventStream = nameTimesPatternStream.select(getNameTimesPatternSelectFunction()); eventStream.print(); env.execute(); } private static Pattern<String, ?> getNameTimesPattern() { AfterMatchSkipStrategy skipStrategy = AfterMatchSkipStrategy.skipPastLastEvent(); return Pattern.<String>begin("begin", skipStrategy) .times(3).within(Time.seconds(5)) .next("next").where(new IterativeCondition<String>() { @Override public boolean filter(String s, Context<String> context) { return "z3".equals(JSONObject.parseObject(s).get("name")); } }) .followedBy("follow").where(new IterativeCondition<String>() { @Override public boolean filter(String s, Context<String> context) { return "z3".equals(JSONObject.parseObject(s).get("name")); } }); } private static PatternStream<String> getNameTimesPatternStream(DataStream<String> dataStream, Pattern<String, ?> nameTimesPattern) { return CEP.pattern(dataStream.keyBy(x -> JSONObject.parseObject(x).get("name")), nameTimesPattern); } private static PatternSelectFunction<String, String> getNameTimesPatternSelectFunction() { return (PatternSelectFunction<String, String>) map -> { Iterator<List<String>> iterator = map.values().iterator(); iterator.next(); return String.format("event happens,name is \"%s\"" , JSONObject.parseObject(iterator.next().get(0)).getString("name")); }; }

测试数据{"id": "1","name": "z3","age": "23","time": 1589015374489}

连续发送五次时可以看到：

 

 

代码中getNameTimesPattern()方法中为定义事件模板，其中next和followBy均为开启下一个模式，但是其代表的严格性不同，主要分为三种：严格连续性（next/notNext），宽松连续性（followedBy/notFollowedBy），和非确定宽松连续性（followedByAny），具体参考下图：

 

 

getNameTimesPatternStream()方法对数据流进行了一定的处理，代码里keyBy将数据转化为json格式并以name项作为key，因此模式中的.times(3)为相同的name出现三次，如果没有这里的keyBy则为相同的数据出现三次，同时也可以在每个模式（每次调用next、followBy、followByAny等开启新模式）中使用.where并结合相关算子进行一些数据处理操作。

 

 

PatternSelectFunction()方法定义了触发模板事件之后的行为，参数可以拿到触发事件的全部数据。

 

参考资料：https://www.cnblogs.com/zhaowei121/p/12060736.html