java flink（二十六）实战之电商黑名单过滤 Flink CEP编程实现、什么是CEP、CEP组合模式demo、CEP循环模式demo

冷不防 2022-10-14 10:52 202阅读 0赞

### **什么是CEP：** ###

**1、复杂事件处理**

**2、Flink中实现复杂事件处理库**

**3、CEP允许在无休止的事件中检测事件模式，让我们有机会掌握数据中的重要部分**

**4、一个或多个由简单事件构成的事件通过一定的规则匹配，然后输出用户想要的数据。**

### **CEP API:** ###

**begin 第一个事件定义 **

**where 条件**

**next 后一个紧跟着事件定义**

**subtype 子类型判断**

**followedBy 后边的时间 不一定紧跟着**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzcxNTY3_size_16_color_FFFFFF_t_70][]

### CEP 个体模式 ： ###

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzcxNTY3_size_16_color_FFFFFF_t_70 1][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzcxNTY3_size_16_color_FFFFFF_t_70 2][]

### **CEP 模式序列**  ###

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzcxNTY3_size_16_color_FFFFFF_t_70 3][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzcxNTY3_size_16_color_FFFFFF_t_70 4][]

### **CEP 超时处理**  ###

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzcxNTY3_size_16_color_FFFFFF_t_70 5][]

### **CEP组合模式demo**  ###

**题目：根据上篇的黑名单过滤功能，我们进行优化，如果两条失败之间穿插了一条乱序的成功登录，那么这两条失败不会被检测。**

**我们利用CEP，首先第一个事件是检测第一条失败登录，然后第二个事件是检测第二个失败登录，进行筛选打印报警信息。**

1、引入jar包：

![20210521105446745.png][]

2、代码展示

package Project;
    
    import Beans.LoginEvent;
    import Beans.LoginFailWarning;
    import org.apache.flink.cep.CEP;
    import org.apache.flink.cep.PatternSelectFunction;
    import org.apache.flink.cep.PatternStream;
    import org.apache.flink.cep.pattern.Pattern;
    import org.apache.flink.cep.pattern.conditions.SimpleCondition;
    import org.apache.flink.streaming.api.TimeCharacteristic;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
    import org.apache.flink.streaming.api.windowing.time.Time;
    
    import java.net.URL;
    import java.util.List;
    import java.util.Map;
    
    public class LoginFailWithCep {
        public static void main(String[] args) throws Exception{
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(1);
            env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
    
            //读取数据
            URL resource = LoginFailWithCep.class.getResource("/LoginLog.csv");
            DataStream<LoginEvent> logEventStream = env.readTextFile(resource.getPath())
                    .map(line -> {
                        String[] fields = line.split(",");
                        return new LoginEvent(new Long(fields[0]), fields[1], fields[2], new Long(fields[3]));
                    })
                    .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<LoginEvent>(Time.seconds(3)) {
                        @Override
                        public long extractTimestamp(LoginEvent loginEvent) {
                            return loginEvent.getTimestamp() * 1000;
                        }
                    });
            //定义匹配模式
            //匹配第一次失败与第二次失败在2s内
            Pattern<LoginEvent, LoginEvent> loginFailPattern = Pattern.<LoginEvent>begin("firstFail").where(new SimpleCondition<LoginEvent>() {
                @Override
                public boolean filter(LoginEvent loginEvent) throws Exception {
                    return "fail".equals(loginEvent.getLoginState());
                }
            }).next("secondFail").where(new SimpleCondition<LoginEvent>() {
                @Override
                public boolean filter(LoginEvent loginEvent) throws Exception {
                    return "fail".equals(loginEvent.getLoginState());
                }
            }).within(Time.seconds(2)); //2s内
            //将匹配模式应用到数据流上
            PatternStream<LoginEvent> patternStream = CEP.pattern(logEventStream.keyBy(LoginEvent::getUserId), loginFailPattern);
            //检出符合匹配条件的复杂事件 进行转换处理 得到报警信息
            SingleOutputStreamOperator<LoginFailWarning> warningStream = patternStream.select(new LoginFailMatchDetectWarning());
            warningStream.print();
            env.execute("login fail detect with cep job");
        }
        //实现自定义的pattern select function
        public static class LoginFailMatchDetectWarning implements PatternSelectFunction<LoginEvent, LoginFailWarning>{
            @Override
            public LoginFailWarning select(Map<String, List<LoginEvent>> map) throws Exception {
                LoginEvent firstFailEvent = map.get("firstFail").get(0);
                LoginEvent secondFailEvent = map.get("secondFail").get(0);
                return new LoginFailWarning(firstFailEvent.getUserId(),firstFailEvent.getTimestamp(),secondFailEvent.getTimestamp(),"login fail 2 times");
            }
        }
    }

**题目二：如果是检测连续三条失败，那么我们还要继续编写组合模式的条件，这样比较麻烦，我们利用CEP循环模式进行实现**

package Project;
    
    import Beans.LoginEvent;
    import Beans.LoginFailWarning;
    import org.apache.flink.cep.CEP;
    import org.apache.flink.cep.PatternSelectFunction;
    import org.apache.flink.cep.PatternStream;
    import org.apache.flink.cep.pattern.Pattern;
    import org.apache.flink.cep.pattern.conditions.SimpleCondition;
    import org.apache.flink.streaming.api.TimeCharacteristic;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
    import org.apache.flink.streaming.api.windowing.time.Time;
    
    import java.net.URL;
    import java.util.List;
    import java.util.Map;
    
    public class LoginFailWithCep {
        public static void main(String[] args) throws Exception{
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(1);
            env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
    
            //读取数据
            URL resource = LoginFailWithCep.class.getResource("/LoginLog.csv");
            DataStream<LoginEvent> logEventStream = env.readTextFile(resource.getPath())
                    .map(line -> {
                        String[] fields = line.split(",");
                        return new LoginEvent(new Long(fields[0]), fields[1], fields[2], new Long(fields[3]));
                    })
                    .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<LoginEvent>(Time.seconds(3)) {
                        @Override
                        public long extractTimestamp(LoginEvent loginEvent) {
                            return loginEvent.getTimestamp() * 1000;
                        }
                    });
            //定义匹配模式
            //匹配第一次失败与第二次失败在2s内
            Pattern<LoginEvent, LoginEvent> loginFailPattern = Pattern.<LoginEvent>begin("failEvents")
                    .where(new SimpleCondition<LoginEvent>() {
                @Override
                public boolean filter(LoginEvent loginEvent) throws Exception {
                    return "fail".equals(loginEvent.getLoginState());
                }})
                    //.times(3) //连续三次  非严格近邻（只要后边有就算）
                    .times(3).consecutive() //必须严格近邻
                    .within(Time.seconds(5));
    
            //将匹配模式应用到数据流上
            PatternStream<LoginEvent> patternStream = CEP.pattern(logEventStream.keyBy(LoginEvent::getUserId), loginFailPattern);
            //检出符合匹配条件的复杂事件 进行转换处理 得到报警信息
            SingleOutputStreamOperator<LoginFailWarning> warningStream = patternStream.select(new LoginFailMatchDetectWarning());
            warningStream.print();
            env.execute("login fail detect with cep job");
        }
        //实现自定义的pattern select function
        public static class LoginFailMatchDetectWarning implements PatternSelectFunction<LoginEvent, LoginFailWarning>{
            @Override
            public LoginFailWarning select(Map<String, List<LoginEvent>> map) throws Exception {
    //            LoginEvent firstFailEvent = map.get("firstFail").get(0);
    //            LoginEvent secondFailEvent = map.get("secondFail").get(0);
    //            return new LoginFailWarning(firstFailEvent.getUserId(),firstFailEvent.getTimestamp(),secondFailEvent.getTimestamp(),"login fail 2 times");
                LoginEvent firstFailEvent = map.get("failEvents").get(0);
                LoginEvent lastFailEvent = map.get("failEvents").get(map.size()-1);
                return new LoginFailWarning(firstFailEvent.getUserId(),firstFailEvent.getTimestamp(),lastFailEvent.getTimestamp(),"login fail 3 times");
            }
        }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzcxNTY3_size_16_color_FFFFFF_t_70]: /images/20221014/09d3a9252ea04e4bb9a47110d3515a52.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzcxNTY3_size_16_color_FFFFFF_t_70 1]: /images/20221014/b62f4c7980094617ad9e9ef8e1323fd9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzcxNTY3_size_16_color_FFFFFF_t_70 2]: /images/20221014/20afc4a23b524d60949d29a1547c9af8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzcxNTY3_size_16_color_FFFFFF_t_70 3]: /images/20221014/69c336b1c5b54fe1a5a8ff8149f38eeb.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzcxNTY3_size_16_color_FFFFFF_t_70 4]: /images/20221014/c8b4c48ad6b94778b6dc0a234d9e833b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzcxNTY3_size_16_color_FFFFFF_t_70 5]: /images/20221014/fcff754a65f84848b6ef4d8c0bb1a595.png
[20210521105446745.png]: /images/20221014/d0e816d34b8241c2ac33fa0f3112d2d7.png