这篇文章给大家介绍FLINK中如何使用进行网站监控报警和报警恢复，内容非常详细，感兴趣的小伙伴们可以参考借鉴，希望对大家能有所帮助。
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
在我们获得了相应的报警和恢复之后，接下来就是调用报警接口进行处理了，我们这只是简单的打印出来信息。
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; } });关于FLINK中如何使用进行网站监控报警和报警恢复就分享到这里了，希望以上内容可以对大家有一定的帮助，可以学到更多知识。如果觉得文章不错，可以把它分享出去让更多的人看到。
亿速云「云服务器」，即开即用、新一代英特尔至强铂金CPU、三副本存储NVMe SSD云盘，价格低至29元/月。点击查看>>
免责声明：本站发布的内容（图片、视频和文字）以原创、转载和分享为主，文章观点不代表本网站立场，如果涉及侵权请联系站长邮箱：is@yisu.com进行举报，并提供相关证据，一经查实，将立刻删除涉嫌侵权内容。
原文链接：https://my.oschina.net/u/4596020/blog/4439393
