FLINK实战-使用CEP进行网站监控报警和报警恢复
flink CEP 简介
flink CEP(Complex event processing),是在Flink之上实现的复杂事件处理库,可以允许我们在不断的流式数据中通过我们自己定义的模式(Pattern)检测和获取出我们想要的数据,然后对这些数据进行下一步的处理。通过各种pattern的组合,我们可以定义出非常复杂的模式来匹配我们的数据。
网上讲CEP原理和用法的文章很多,大家可以参考下 https://juejin.im/post/5de1f32af265da05cc3190f9#heading-9
简单来说一下,其实我们可以把使用flink cep当做我们平时用的正则表达式,cep中的Pattern就是我们定义的正则表达式,flink中的DataStream就是正则表达式中待匹配的字符串,flink 通过DataStream 和 自定义的Pattern进行匹配,生成一个经过过滤之后的DataStream .
基于自定义的pattern,我们可以做很多工作,比如监控报警、风控、反爬等等,接下来我们基于一个简单的报警小例子来讲解一些FLINK cep的实际应用。
案例详解
我们基于flink CEP做一个简单的报警,首先我们简化一下报警的需求 1.统计出来每秒钟http状态码为非200的数量所占比例。大于0.7的时候触发报警。 2.统计结果连续发生三大于阈值(0.7,这个数字是我自己写的,为了测试用,真实环境需要根据实际经验来设置)发送报警通知。 3.统计结果小于等于阈值触发报警恢复通知。
实际应用中我们一般会去消费kafka的数据来作为source、这里我们为了简化,通过自定义source生成一些模拟的数据。
public static class MySource implements SourceFunction<Tuple4<String,Long,Integer,Integer>>{
static int status[] = {200, 404, 500, 501, 301};
@Override
public void run(SourceContext<Tuple4<String,Long,Integer,Integer>> sourceContext) throws Exception{
while (true){
Thread.sleep((int) (Math.random() * 100));
// traceid,timestamp,status,response time
Tuple4 log = Tuple4.of(
UUID.randomUUID().toString(),
System.currentTimeMillis(),
status[(int) (Math.random() * 4)],
(int) (Math.random() * 100));
sourceContext.collect(log);
}
}
@Override
public void cancel(){
}
}
接下来我们定义一个sql,用来计算我们的需求中的第一个要求。
String sql = "select pv,errorcount,round(CAST(errorcount AS DOUBLE)/pv,2) as errorRate," +
"(starttime + interval '8' hour ) as stime," +
"(endtime + interval '8' hour ) as etime " +
"from (select count(*) as pv," +
"sum(case when status = 200 then 0 else 1 end) as errorcount, " +
"TUMBLE_START(proctime,INTERVAL '1' SECOND) as starttime," +
"TUMBLE_END(proctime,INTERVAL '1' SECOND) as endtime " +
"from log group by TUMBLE(proctime,INTERVAL '1' SECOND) )";
通过执行sql,我们获取到了一个Result对象的DataStream,
Table table = tenv.sqlQuery(sql);
DataStream<Result> ds1 = tenv.toAppendStream(table, Result.class);
接下来我们到了最核心的地方,我们需要定一个Pattern。
Pattern pattern = Pattern.<Result>begin("alert").where(new IterativeCondition<Result>(){
@Override
public boolean filter(
Result i, Context<Result> context) throws Exception{
return i.getErrorRate() > 0.7D;
}
}).times(3).consecutive().followedBy("recovery").where(new IterativeCondition<Result>(){
@Override
public boolean filter(
Result i,
Context<Result> context) throws Exception{
return i.getErrorRate() <= 0.7D;
}
}).optional();
来详细解释一下这个Pattern
- 首先定义一个名为alert的Pattern,该Pattern的作用就是过滤出错误率大于0.7的数据,
- times(3) ,表示要匹配三次,也就是要三次大于0.7.
- consecutive 表示上述的三次匹配要是连续的,比如0.75、0.8、0.78,只有类似这样的数据才能被匹配到,中间不能有不符合的数据出现。
- followedBy表示该alert pattern的下面要跟着一个recovery pattern,而followedBy是宽松匹配,也就是两个模式之间可以有其他的数据,如果要采用严格匹配,是使用next.
- 最后recovery pattern加上一个optional 是我为了区分报警,和报警恢复想的的一个方案,这样的话,如果是只匹配到了alert pattern,输出的就是报警,如果recovery pattern也匹配到了,那么就是报警恢复。
在我们获得了相应的报警和恢复之后,接下来就是调用报警接口进行处理了,我们这只是简单的打印出来信息。
DataStream<Map<String,List<Result>>> alertStream = org.apache.flink.cep.CEP.pattern(
ds1,
pattern).select(new PatternSelectFunction<Result,Map<String,List<Result>>>(){
@Override
public Map<String,List<Result>> select(Map<String,List<Result>> map) throws Exception{
List<Result> alertList = map.get("alert");
List<Result> recoveryList = map.get("recovery");
if (recoveryList != null){
System.out.print("接受到了报警恢复的信息,报警信息如下:");
System.out.print(alertList);
System.out.print(" 对应的恢复信息:");
System.out.println(recoveryList);
} else {
System.out.print("收到了报警信息 ");
System.out.print(alertList);
}
return map;
}
});
完整的代码请参考 https://github.com/zhangjun0x01/bigdata-examples/blob/master/flink/src/main/java/cep/WebMonitorAlert.java
- JavaScript 教程
- JavaScript 编辑工具
- JavaScript 与HTML
- JavaScript 与Java
- JavaScript 数据结构
- JavaScript 基本数据类型
- JavaScript 特殊数据类型
- JavaScript 运算符
- JavaScript typeof 运算符
- JavaScript 表达式
- JavaScript 类型转换
- JavaScript 基本语法
- JavaScript 注释
- Javascript 基本处理流程
- Javascript 选择结构
- Javascript if 语句
- Javascript if 语句的嵌套
- Javascript switch 语句
- Javascript 循环结构
- Javascript 循环结构实例
- Javascript 跳转语句
- Javascript 控制语句总结
- Javascript 函数介绍
- Javascript 函数的定义
- Javascript 函数调用
- Javascript 几种特殊的函数
- JavaScript 内置函数简介
- Javascript eval() 函数
- Javascript isFinite() 函数
- Javascript isNaN() 函数
- parseInt() 与 parseFloat()
- escape() 与 unescape()
- Javascript 字符串介绍
- Javascript length属性
- javascript 字符串函数
- Javascript 日期对象简介
- Javascript 日期对象用途
- Date 对象属性和方法
- Javascript 数组是什么
- Javascript 创建数组
- Javascript 数组赋值与取值
- Javascript 数组属性和方法
- CentOS7.5安装配置Harbor1.7的全过程
- Linux CentOS 定时运行脚本配置的方法
- 从Centos7升级到Centos8的教程(图文详解)
- Linux(Centos7)下redis5集群搭建和使用说明详解
- CentOS7下安装yum源及上传下载命令rz、sz安装方法(图解)
- C#实例:四路激光测距雷达数据采集和波形图绘制
- Linux 中有效用户组和初始用户组的实现
- ubuntu 16.04 64位兼容32位程序三步曲
- crontab执行结果未通过发送mail通知用户的方法
- 如何将CentOS7升级至CentOS8(详细步骤)
- leetcode树之二叉树的所有路径
- 如何利用Bash脚本监控Linux的内存使用情况
- Ubuntu18 给terminal改个漂亮的命令行提示符的方法
- leetcode树之将有序数组转换为二叉搜索树
- 在 Ubuntu 上安装 Protobuf 3 的教程详解