第一章.Flink CEP介绍

1.什么是FlinkCEP

  1.     FlinkCEP是在Flink实现的复杂事件处理库,它可以让你在无界流中检测出特定的数据,有机会掌握数据中重要的那部分
  2.     是一种基于动态环境中事件流的分析技术,事件在这里通常是有意义的状态变化,通过分析事件间的关系,利用过滤,联合聚合等技术,根据事件间的聚合关系和时序关系制定检测规则,持续的从事件流中查询出符合要求的事件序列,最终分析得到更复杂的复合事件
  3.     1.目标:从有序的简单事件流中发现一些高阶特征
  4.     2.输入:一个或多个由简单事件构成的事件流
  5.     3.处理:识别简单事件之间的内在联系,多个复合一定规则的简单事件构成复杂事件
  6.     4.输出:满足规则的复杂事件
  1.     FlinkCEP应用场景
  2.     1.风险控制
  3.     2.策略营销
  4.     3.运维监控

2.Flink CEP开发

一.导入相关依赖

  1. <dependency>
  2.     <groupId>org.apache.flink</groupId>
  3.     <artifactId>flink-cep_${scala.binary.version}</artifactId>
  4.     <version>${flink.version}</version>
  5. </dependency>

二.基本使用

  1. package com.atguigu.flink.day08;
  3. import com.atguigu.flink.day02.pojo.WaterSensor;
  4. import org.apache.flink.api.common.eventtime.WatermarkStrategy;
  5. import org.apache.flink.cep.CEP;
  6. import org.apache.flink.cep.PatternSelectFunction;
  7. import org.apache.flink.cep.PatternStream;
  8. import org.apache.flink.cep.nfa.aftermatch.AfterMatchSkipStrategy;
  9. import org.apache.flink.cep.pattern.Pattern;
  10. import org.apache.flink.cep.pattern.conditions.SimpleCondition;
  11. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  12. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  14. import java.time.Duration;
  15. import java.util.List;
  16. import java.util.Map;
  18. public class $01_CEPBaseUse {
  19.     public static void main(String[] args) throws Exception {
  20.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  21.         env.setParallelism(1);
  22.         //1.先有数据流
  23.         SingleOutputStreamOperator<WaterSensor> stream = env.readTextFile("input/sensor.txt")
  24.                 .map(line -> {
  25.                     String[] data = line.split(",");
  26.                     return new WaterSensor(
  27.                             data[0],
  28.                             Long.parseLong(data[1]),
  29.                             Integer.parseInt(data[2])
  30.                     );
  31.                 })
  32.                 .assignTimestampsAndWatermarks(
  33.                         WatermarkStrategy
  34.                                 .<WaterSensor>forMonotonousTimestamps()
  35.                                 .withTimestampAssigner((ws, ts) -> ws.getTs())
  37.                 );
  38.         //2.定义模式
  39.         Pattern<WaterSensor, WaterSensor> pattern = Pattern.<WaterSensor>begin("start")
  40.                 .where(new SimpleCondition<WaterSensor>() {
  41.                     @Override
  42.                     public boolean filter(WaterSensor value) throws Exception {
  43.                         return "sensor_1".equals(value.getId());
  44.                     }
  45.                 });
  46.         //3.把模式作用在数据流上
  47.         PatternStream<WaterSensor> ps = CEP.pattern(stream, pattern);
  48.         //4..获取匹配到的数据
  49.         ps.select(new PatternSelectFunction<WaterSensor, String>() {
  50.             @Override
  51.             public String select(Map<String, List<WaterSensor>> map) throws Exception {
  52.                 return map.toString();
  53.             }
  54.         }).print();
  55.         env.execute();
  56.     }
  57. }

三.模式API

  1.     模式API可以让你定义想从输入流中抽取的复杂模式序列
  2.     模式:比如找拥有相同属性事件序列的模式
  3.         注意:1.每个模式必须有一个独一无二的名字,你可以在后面使用它来识别匹配到的事件
  4.             2.模式的名字不能包含字符":"
  5.     模式序列:每个复杂的模式序列包括多个简单的模式,也叫模式序列,你可以把模式序列看做是这样的模式构成的图,这些模式基于用户指定的条件从一个转换到另外一个
  6.     匹配:输入事件的一个序列,这些序列通过一系列有效的模式转换,能够访问到复杂模式图中的所有模式

1.单例模式

单例模式只接受一个事件,默认情况模式都是单例模式

上面的例子就是一个单例模式

2.组合模式

把多个单个模式组合在一起就是组合模式,组合模式由一个初始化模式(.begin(…))开头

  1. package com.atguigu.flink.day08;
  3. import com.atguigu.flink.day02.pojo.WaterSensor;
  4. import org.apache.flink.api.common.eventtime.WatermarkStrategy;
  5. import org.apache.flink.cep.CEP;
  6. import org.apache.flink.cep.PatternSelectFunction;
  7. import org.apache.flink.cep.PatternStream;
  8. import org.apache.flink.cep.pattern.Pattern;
  9. import org.apache.flink.cep.pattern.conditions.SimpleCondition;
  10. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  11. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  13. import java.util.List;
  14. import java.util.Map;
  16. public class $04_CEPBaseUseCombine {
  17.     public static void main(String[] args) throws Exception {
  18.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  19.         env.setParallelism(1);
  20.         //1.先有数据流
  21.         SingleOutputStreamOperator<WaterSensor> stream = env.readTextFile("input/sensor.txt")
  22.                 .map(line -> {
  23.                     String[] data = line.split(",");
  24.                     return new WaterSensor(
  25.                             data[0],
  26.                             Long.parseLong(data[1]),
  27.                             Integer.parseInt(data[2])
  28.                     );
  29.                 })
  30.                 .assignTimestampsAndWatermarks(
  31.                         WatermarkStrategy
  32.                                 .<WaterSensor>forMonotonousTimestamps()
  33.                                 .withTimestampAssigner((ws, ts) -> ws.getTs())
  35.                 );
  36.         //2.定义模式
  37.         Pattern<WaterSensor, WaterSensor> pattern = Pattern.<WaterSensor>begin("start")
  38.                 .where(new SimpleCondition<WaterSensor>() {
  39.                     @Override
  40.                     public boolean filter(WaterSensor value) throws Exception {
  41.                         return "sensor_1".equals(value.getId());
  42.                     }
  43.                 })
  44.                 //.next("end") 严格连续
  45.                 //.followedBy("end") 松散连续
  46.                 .followedByAny("end") //不确定的松散连续
  47.                 .where(new SimpleCondition<WaterSensor>() {
  48.                     @Override
  49.                     public boolean filter(WaterSensor value) throws Exception {
  50.                         return "sensor_2".equals(value.getId());
  51.                     }
  52.                 });
  53.         //3.把模式作用在数据流上
  54.         PatternStream<WaterSensor> ps = CEP.pattern(stream, pattern);
  55.         //4..获取匹配到的数据
  56.         ps.select(new PatternSelectFunction<WaterSensor, String>() {
  57.             @Override
  58.             public String select(Map<String, List<WaterSensor>> map) throws Exception {
  59.                 return map.toString();
  60.             }
  61.         }).print();
  62.         env.execute();
  63.     }
  64. }

3.循环模式

循环模式可以接收多个事件

单例模式配合上量词就是循环模式(非常类似正则表达式)

  1. package com.atguigu.flink.day08;
  3. import com.atguigu.flink.day02.pojo.WaterSensor;
  4. import org.apache.flink.api.common.eventtime.WatermarkStrategy;
  5. import org.apache.flink.cep.CEP;
  6. import org.apache.flink.cep.PatternSelectFunction;
  7. import org.apache.flink.cep.PatternStream;
  8. import org.apache.flink.cep.pattern.Pattern;
  9. import org.apache.flink.cep.pattern.conditions.SimpleCondition;
  10. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  11. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  13. import java.util.List;
  14. import java.util.Map;
  16. public class $02_CEPBaseUseLoop {
  17.     public static void main(String[] args) throws Exception {
  18.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  19.         env.setParallelism(1);
  20.         //1.先有数据流
  21.         SingleOutputStreamOperator<WaterSensor> stream = env.readTextFile("input/sensor.txt")
  22.                 .map(line -> {
  23.                     String[] data = line.split(",");
  24.                     return new WaterSensor(
  25.                             data[0],
  26.                             Long.parseLong(data[1]),
  27.                             Integer.parseInt(data[2])
  28.                     );
  29.                 })
  30.                 .assignTimestampsAndWatermarks(
  31.                         WatermarkStrategy
  32.                                 .<WaterSensor>forMonotonousTimestamps()
  33.                                 .withTimestampAssigner((ws, ts) -> ws.getTs())
  35.                 );
  36.         //2.定义模式
  37.         Pattern<WaterSensor, WaterSensor> pattern = Pattern.<WaterSensor>begin("start")
  38.                 .where(new SimpleCondition<WaterSensor>() {
  39.                     @Override
  40.                     public boolean filter(WaterSensor value) throws Exception {
  41.                         return "sensor_1".equals(value.getId());
  42.                     }
  43.                 })
  44.                 //.times(2) 出现两次;
  45.                 //.oneOrMore() //一次或多次
  46.                 //.times(2,4) //大于等于两次小于四次
  47.                 .timesOrMore(2)//2次或2次以上
  48.         .until(new SimpleCondition<WaterSensor>() {
  49.             @Override
  50.             public boolean filter(WaterSensor value) throws Exception {
  51.                 return "sensor_2".equals(value.getId());
  52.             }
  53.         });
  54.         //3.把模式作用在数据流上
  55.         PatternStream<WaterSensor> ps = CEP.pattern(stream, pattern);
  56.         //4..获取匹配到的数据
  57.         ps.select(new PatternSelectFunction<WaterSensor, String>() {
  58.             @Override
  59.             public String select(Map<String, List<WaterSensor>> map) throws Exception {
  60.                 return map.toString();
  61.             }
  62.         }).print();
  63.         env.execute();
  64.     }
  65. }

4.循环模式的连续性

松散连续(默认)

忽略匹配的事件之间的不匹配的事件

  1. Pattern<WaterSensor, WaterSensor> pattern = Pattern.<WaterSensor>begin("start")
  2.                 .where(new SimpleCondition<WaterSensor>() {
  3.                     @Override
  4.                     public boolean filter(WaterSensor value) throws Exception {
  5.                         return "sensor_1".equals(value.getId());
  6.                     }
  7.                 })
  8.                 .times(2); //出现两次

严格连续

期望所有匹配的事件严格的一个接一个出现,中间没有任何不匹配的事件

  1. Pattern<WaterSensor, WaterSensor> pattern = Pattern.<WaterSensor>begin("start")
  2.                 .where(new SimpleCondition<WaterSensor>() {
  3.                     @Override
  4.                     public boolean filter(WaterSensor value) throws Exception {
  5.                         return "sensor_1".equals(value.getId());
  6.                     }
  7.                 })
  8.                 .times(2) //出现两次
  9.                 .consecutive();

非确定的松散连续

更近一步的松散连续,允许忽略掉一些匹配事件的附加匹配

  1. Pattern<WaterSensor, WaterSensor> pattern = Pattern.<WaterSensor>begin("start")
  2.                 .where(new SimpleCondition<WaterSensor>() {
  3.                     @Override
  4.                     public boolean filter(WaterSensor value) throws Exception {
  5.                         return "sensor_1".equals(value.getId());
  6.                     }
  7.                 })
  8.                 .times(2) //出现两次
  9.                 .allowCombinations();

5.循环模式的贪婪性

在组合模式的情况下,对次数的处理尽可能的获取最多的那个个数,就是贪婪,当一个事件同时满足两个模式的时候起作用

  1. package com.atguigu.flink.day08;
  3. import com.atguigu.flink.day02.pojo.WaterSensor;
  4. import org.apache.flink.api.common.eventtime.WatermarkStrategy;
  5. import org.apache.flink.cep.CEP;
  6. import org.apache.flink.cep.PatternSelectFunction;
  7. import org.apache.flink.cep.PatternStream;
  8. import org.apache.flink.cep.pattern.Pattern;
  9. import org.apache.flink.cep.pattern.conditions.SimpleCondition;
  10. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  11. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  13. import java.util.List;
  14. import java.util.Map;
  16. public class $03_CEPBaseUseGreedy {
  17.     public static void main(String[] args) throws Exception {
  18.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  19.         env.setParallelism(1);
  20.         //1.先有数据流
  21.         SingleOutputStreamOperator<WaterSensor> stream = env.readTextFile("input/sensor.txt")
  22.                 .map(line -> {
  23.                     String[] data = line.split(",");
  24.                     return new WaterSensor(
  25.                             data[0],
  26.                             Long.parseLong(data[1]),
  27.                             Integer.parseInt(data[2])
  28.                     );
  29.                 })
  30.                 .assignTimestampsAndWatermarks(
  31.                         WatermarkStrategy
  32.                                 .<WaterSensor>forMonotonousTimestamps()
  33.                                 .withTimestampAssigner((ws, ts) -> ws.getTs())
  35.                 );
  36.         //2.定义模式
  37.         Pattern<WaterSensor, WaterSensor> pattern = Pattern.<WaterSensor>begin("start")
  38.                 .where(new SimpleCondition<WaterSensor>() {
  39.                     @Override
  40.                     public boolean filter(WaterSensor value) throws Exception {
  41.                         return "sensor_1".equals(value.getId());
  42.                     }
  43.                 }).times(2,3).greedy()
  44.                 .next("end")
  45.                 .where(new SimpleCondition<WaterSensor>() {
  46.                     @Override
  47.                     public boolean filter(WaterSensor value) throws Exception {
  48.                         return value.getVc() == 30;
  49.                     }
  50.                 });
  51.         //3.把模式作用在数据流上
  52.         PatternStream<WaterSensor> ps = CEP.pattern(stream, pattern);
  53.         //4..获取匹配到的数据
  54.         ps.select(new PatternSelectFunction<WaterSensor, String>() {
  55.             @Override
  56.             public String select(Map<String, List<WaterSensor>> map) throws Exception {
  57.                 return map.toString();
  58.             }
  59.         }).print();
  60.         env.execute();
  61.     }
  62. }

6.模式的可选性

pattern.optional()方法让所有的模式变成可选的,不管是否是循环模式

  1. package com.atguigu.flink.day08;
  3. import com.atguigu.flink.day02.pojo.WaterSensor;
  4. import org.apache.flink.api.common.eventtime.WatermarkStrategy;
  5. import org.apache.flink.cep.CEP;
  6. import org.apache.flink.cep.PatternSelectFunction;
  7. import org.apache.flink.cep.PatternStream;
  8. import org.apache.flink.cep.pattern.Pattern;
  9. import org.apache.flink.cep.pattern.conditions.SimpleCondition;
  10. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  11. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  13. import java.util.List;
  14. import java.util.Map;
  16. public class $05_CEPBaseUseOptional {
  17.     public static void main(String[] args) throws Exception {
  18.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  19.         env.setParallelism(1);
  20.         //1.先有数据流
  21.         SingleOutputStreamOperator<WaterSensor> stream = env.readTextFile("input/sensor.txt")
  22.                 .map(line -> {
  23.                     String[] data = line.split(",");
  24.                     return new WaterSensor(
  25.                             data[0],
  26.                             Long.parseLong(data[1]),
  27.                             Integer.parseInt(data[2])
  28.                     );
  29.                 })
  30.                 .assignTimestampsAndWatermarks(
  31.                         WatermarkStrategy
  32.                                 .<WaterSensor>forMonotonousTimestamps()
  33.                                 .withTimestampAssigner((ws, ts) -> ws.getTs())
  35.                 );
  36.         //2.定义模式
  37.         Pattern<WaterSensor, WaterSensor> pattern = Pattern.<WaterSensor>begin("start")
  38.                 .where(new SimpleCondition<WaterSensor>() {
  39.                     @Override
  40.                     public boolean filter(WaterSensor value) throws Exception {
  41.                         return "sensor_1".equals(value.getId());
  42.                     }
  43.                 }).times(2).optional() //出现0次或2次
  44.                 .next("end")
  45.                 .where(new SimpleCondition<WaterSensor>() {
  46.                     @Override
  47.                     public boolean filter(WaterSensor value) throws Exception {
  48.                         return "sensor_2".equals(value.getId());
  49.                     }
  50.                 });
  51.         //3.把模式作用在数据流上
  52.         PatternStream<WaterSensor> ps = CEP.pattern(stream, pattern);
  53.         //4..获取匹配到的数据
  54.         ps.select(new PatternSelectFunction<WaterSensor, String>() {
  55.             @Override
  56.             public String select(Map<String, List<WaterSensor>> map) throws Exception {
  57.                 return map.toString();
  58.             }
  59.         }).print();
  60.         env.execute();
  61.     }
  62. }

7.模式组

  1. 在前面的代码中次数只能用在某个模式上, 比如: .begin(...).where(...).next(...).where(...).times(2)  这里的次数只会用在next这个模式上, 而不会用在begin模式上.<br />    如果需要用在多个模式上,可以使用模式组!
  1. package com.atguigu.flink.day08;
  3. import com.atguigu.flink.day02.pojo.WaterSensor;
  4. import org.apache.flink.api.common.eventtime.WatermarkStrategy;
  5. import org.apache.flink.cep.CEP;
  6. import org.apache.flink.cep.PatternSelectFunction;
  7. import org.apache.flink.cep.PatternStream;
  8. import org.apache.flink.cep.pattern.Pattern;
  9. import org.apache.flink.cep.pattern.conditions.SimpleCondition;
  10. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  11. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  13. import java.util.List;
  14. import java.util.Map;
  16. public class $06_CEPBaseUseGroup {
  17.     public static void main(String[] args) throws Exception {
  18.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  19.         env.setParallelism(1);
  20.         //1.先有数据流
  21.         SingleOutputStreamOperator<WaterSensor> stream = env.readTextFile("input/sensor.txt")
  22.                 .map(line -> {
  23.                     String[] data = line.split(",");
  24.                     return new WaterSensor(
  25.                             data[0],
  26.                             Long.parseLong(data[1]),
  27.                             Integer.parseInt(data[2])
  28.                     );
  29.                 })
  30.                 .assignTimestampsAndWatermarks(
  31.                         WatermarkStrategy
  32.                                 .<WaterSensor>forMonotonousTimestamps()
  33.                                 .withTimestampAssigner((ws, ts) -> ws.getTs())
  35.                 );
  36.         //2.定义模式
  37.         Pattern<WaterSensor, WaterSensor> pattern = Pattern.begin(Pattern.
  38.                 <WaterSensor>begin("start")
  39.                 .where(new SimpleCondition<WaterSensor>() {
  40.                     @Override
  41.                     public boolean filter(WaterSensor value) throws Exception {
  42.                         return "sensor_1".equals(value.getId());
  43.                     }
  44.                 })
  45.                 .next("next")
  46.                 .where(new SimpleCondition<WaterSensor>() {
  47.                     @Override
  48.                     public boolean filter(WaterSensor value) throws Exception {
  49.                         return "sensor_2".equals(value.getId());
  50.                     }
  51.                 }))
  52.                 .oneOrMore();
  54.         //3.把模式作用在数据流上
  55.         PatternStream<WaterSensor> ps = CEP.pattern(stream, pattern);
  56.         //4..获取匹配到的数据
  57.         ps.select(new PatternSelectFunction<WaterSensor, String>() {
  58.             @Override
  59.             public String select(Map<String, List<WaterSensor>> map) throws Exception {
  60.                 return map.toString();
  61.             }
  62.         }).print();
  63.         env.execute();
  64.     }
  65. }

8.超时数据

当一个模式上通过within加上窗口长度后，部分匹配的事件序列就可能因为超过窗口长度而被丢弃。

  1. package com.atguigu.flink.day08;
  3. import com.atguigu.flink.day02.pojo.WaterSensor;
  4. import org.apache.flink.api.common.eventtime.WatermarkStrategy;
  5. import org.apache.flink.cep.CEP;
  6. import org.apache.flink.cep.PatternSelectFunction;
  7. import org.apache.flink.cep.PatternStream;
  8. import org.apache.flink.cep.pattern.Pattern;
  9. import org.apache.flink.cep.pattern.conditions.SimpleCondition;
  10. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  11. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  12. import org.apache.flink.streaming.api.windowing.time.Time;
  14. import java.util.List;
  15. import java.util.Map;
  17. public class $07_CEPBaseUseWithIn {
  18.     public static void main(String[] args) throws Exception {
  19.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  20.         env.setParallelism(1);
  21.         //1.先有数据流
  22.         SingleOutputStreamOperator<WaterSensor> stream = env.readTextFile("input/sensor.txt")
  23.                 .map(line -> {
  24.                     String[] data = line.split(",");
  25.                     return new WaterSensor(
  26.                             data[0],
  27.                             Long.parseLong(data[1]),
  28.                             Integer.parseInt(data[2])
  29.                     );
  30.                 })
  31.                 .assignTimestampsAndWatermarks(
  32.                         WatermarkStrategy
  33.                                 .<WaterSensor>forMonotonousTimestamps()
  34.                                 .withTimestampAssigner((ws, ts) -> ws.getTs())
  36.                 );
  37.         //2.定义模式
  38.         Pattern<WaterSensor, WaterSensor> pattern = Pattern.<WaterSensor>begin("start")
  39.                 .where(new SimpleCondition<WaterSensor>() {
  40.                     @Override
  41.                     public boolean filter(WaterSensor value) throws Exception {
  42.                         return "sensor_1".equals(value.getId());
  43.                     }
  44.                 })
  45.                 .next("end")
  46.                 .where(new SimpleCondition<WaterSensor>() {
  47.                     @Override
  48.                     public boolean filter(WaterSensor value) throws Exception {
  49.                         return "sensor_2".equals(value.getId());
  50.                     }
  51.                 })
  52.                 .within(Time.seconds(2));
  53.         //3.把模式作用在数据流上
  54.         PatternStream<WaterSensor> ps = CEP.pattern(stream, pattern);
  55.         //4..获取匹配到的数据
  56.         ps.select(new PatternSelectFunction<WaterSensor, String>() {
  57.             @Override
  58.             public String select(Map<String, List<WaterSensor>> map) throws Exception {
  59.                 return map.toString();
  60.             }
  61.         }).print();
  62.         env.execute();
  63.     }
  64. }

9.匹配后跳过策略

day08[Flink CEP编程] - 图1

第二章.Flink CEP实战

1.恶意登录监控

  1. package com.atguigu.flink.day08;
  3. import com.atguigu.flink.day07.pojo.LoginEvent;
  4. import org.apache.flink.api.common.eventtime.WatermarkStrategy;
  5. import org.apache.flink.cep.CEP;
  6. import org.apache.flink.cep.PatternSelectFunction;
  7. import org.apache.flink.cep.PatternStream;
  8. import org.apache.flink.cep.pattern.Pattern;
  9. import org.apache.flink.cep.pattern.conditions.SimpleCondition;
  10. import org.apache.flink.streaming.api.datastream.KeyedStream;
  11. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  13. import org.apache.flink.streaming.api.windowing.time.Time;
  17. import java.time.Duration;
  19. import java.util.List;
  20. import java.util.Map;
  22. /**
  23.  * 需求:恶意登录监控(同一用户在2s之内连续两次登录失败)
  24.  */
  26. public class $08_Login {
  27.     public static void main(String[] args) throws Exception {
  28.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  29.         env.setParallelism(1);
  30.         KeyedStream<LoginEvent, Long> stream = env.readTextFile("input/LoginLog.csv")
  31.                 .map(line -> {
  32.                     String[] data = line.split(",");
  33.                     return new LoginEvent(
  34.                             Long.parseLong(data[0]),
  35.                             data[1],
  36.                             data[2],
  37.                             Long.parseLong(data[3]) * 1000L
  38.                     );
  39.                 })
  40.                 .assignTimestampsAndWatermarks(
  41.                         WatermarkStrategy.<LoginEvent>forBoundedOutOfOrderness(Duration.ofSeconds(3))
  42.                         .withTimestampAssigner((log,ts)-> log.getEventTime())
  43.                 )
  44.                 .keyBy(log -> log.getUserId());
  45.         //定义模式  连续两秒内登陆失败
  46.         Pattern<LoginEvent, LoginEvent> pattern = Pattern
  47.                 .<LoginEvent>begin("fail_1")
  48.                 .where(new SimpleCondition<LoginEvent>() {
  49.                     @Override
  50.                     public boolean filter(LoginEvent value) throws Exception {
  51.                         return "fail".equals(value.getEventType());
  52.                     }
  53.                 })
  54.                 .next("fail_2")
  55.                 .where(new SimpleCondition<LoginEvent>() {
  56.                     @Override
  57.                     public boolean filter(LoginEvent value) throws Exception {
  58.                         return "fail".equals(value.getEventType());
  59.                     }
  60.                 })
  61.                 .within(Time.seconds(2001));
  62.         //把模式作用在流上
  63.         PatternStream<LoginEvent> ps = CEP.pattern(stream, pattern);
  64.         //获取匹配到的数据
  65.         ps.select(new PatternSelectFunction<LoginEvent, String>() {
  66.             @Override
  67.             public String select(Map<String, List<LoginEvent>> map) throws Exception {
  68.                 LoginEvent fail_1 = map.get("fail_1").get(0);
  69.                 return fail_1.getUserId() + "正在恶意登录,请注意";
  70.             }
  71.         })
  72.                 .print();
  73.         env.execute();
  78.     }
  79. }

2.订单支付实时监控

  1. package com.atguigu.flink.day08;
  3. import com.atguigu.flink.day07.pojo.OrderEvent;
  4. import org.apache.flink.api.common.eventtime.WatermarkStrategy;
  5. import org.apache.flink.cep.CEP;
  6. import org.apache.flink.cep.PatternSelectFunction;
  7. import org.apache.flink.cep.PatternStream;
  8. import org.apache.flink.cep.PatternTimeoutFunction;
  9. import org.apache.flink.cep.nfa.aftermatch.AfterMatchSkipStrategy;
  10. import org.apache.flink.cep.pattern.Pattern;
  11. import org.apache.flink.cep.pattern.conditions.SimpleCondition;
  12. import org.apache.flink.streaming.api.datastream.KeyedStream;
  13. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  14. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  15. import org.apache.flink.streaming.api.windowing.time.Time;
  17. import org.apache.flink.util.OutputTag;
  19. import java.time.Duration;
  20. import java.util.List;
  21. import java.util.Map;
  23. /**
  24.  * 需求:订单支付实时监控
  25.  */
  27. public class $09_Order {
  28.     public static void main(String[] args) throws Exception {
  29.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  30.         env.setParallelism(1);
  31.         KeyedStream<OrderEvent, Long> stream = env.readTextFile("input/OrderLog.csv")
  32.                 .map(line -> {
  33.                     String[] data = line.split(",");
  34.                     return new OrderEvent(
  35.                             Long.valueOf(data[0]),
  36.                             data[1],
  37.                             data[2],
  38.                             Long.parseLong(data[3]) * 1000
  39.                     );
  40.                 })
  41.                 .assignTimestampsAndWatermarks(
  42.                         WatermarkStrategy
  43.                                 .<OrderEvent>forBoundedOutOfOrderness(Duration.ofSeconds(3))
  44.                                 .withTimestampAssigner((event, ts) -> event.getEventTime())
  45.                 )
  46.                 .keyBy(orderEvent -> orderEvent.getOrderId());
  47.         Pattern<OrderEvent, OrderEvent> pattern = Pattern.<OrderEvent>begin("create", AfterMatchSkipStrategy.skipPastLastEvent())
  48.                 .where(new SimpleCondition<OrderEvent>() {
  49.                     @Override
  50.                     public boolean filter(OrderEvent value) throws Exception {
  51.                         return "create".equals(value.getEventType());
  52.                     }
  53.                 }).optional()
  54.                 .next("pay")
  55.                 .where(new SimpleCondition<OrderEvent>() {
  56.                     @Override
  57.                     public boolean filter(OrderEvent value) throws Exception {
  58.                         return "pay".equals(value.getEventType());
  59.                     }
  60.                 })
  61.                 .within(Time.minutes(45));
  62.         PatternStream<OrderEvent> ps = CEP.pattern(stream, pattern);
  63.         SingleOutputStreamOperator<String> normal = ps.select(
  64.                 new OutputTag<String>("late") {
  65.                 },
  66.                 new PatternTimeoutFunction<OrderEvent, String>() {
  67.                     @Override
  68.                     public String timeout(Map<String, List<OrderEvent>> map, long l) throws Exception {
  69.                         OrderEvent create = map.get("create").get(0);
  70.                         return "订单:" + create.getOrderId() + "只有create,没有pay,或者pay超时支付";
  71.                     }
  72.                 },
  73.                 new PatternSelectFunction<OrderEvent, String>() {
  74.                     @Override
  75.                     public String select(Map<String, List<OrderEvent>> map) throws Exception {
  76.                         OrderEvent pay = map.get("pay").get(0);
  77.                         if (!map.containsKey("create")) {
  78.                             return "订单:" + pay.getOrderId() + "只有pay,没有create,系统bug";
  79.                         } else {
  80.                             return "";
  81.                         }
  83.                     }
  84.                 }
  85.         );
  86.         normal.filter(s->s.length()>0)
  87.                 .union(normal.getSideOutput(new OutputTag<String>("late"){}))
  88.                 .print();
  90.         env.execute();
  93.     }
  94. }