- 简介
- 使用方法
- 简述一下需求
- 自定义source
背景
紧接着上一篇文章【FLINK实战-使用CEP进行网站监控报警和报警恢复】，上一篇主要讲了怎么对数据流做监控报警，但是实际生产环境中，可能我们对这个报警的阈值设置成多大也没有一个准，可能需要根据经验不断的来修改，所以就涉及了可能需要不断的修改这个报警的阈值，但是如果每次修改了之后，都通过重启flink程序来实现，这个成本就有点高了，所以我们这次主要是讲解一下，如何使用flink的广播动态的更新配置来设置这个报警的阈值.
flink broadstate
简介
是flink提供的一种算子，可以使用一个Stream接收不断变化的数据（比如我们的配置数据），然后把这些数据广播到flink的所有task中，这样主Stream中的数据就能动态的从广播流中获取所需要的配置，然后根据动态的配置来处理数据.
使用方法
构造广播流
代码语言：javascript复制
//构造一个map state descriptor
MapStateDescriptor<String,Long> confDescriptor = new MapStateDescriptor<>(
"config-keywords",
BasicTypeInfo.STRING_TYPE_INFO,
BasicTypeInfo.LONG_TYPE_INFO);
//配置流，比如我们可以从kafka动态接受配置，或者循环去读取数据库之类的
DataStream confStream = env.addSource(new BroadcastSource());
//构造广播流
BroadcastStream<Integer> broadcastStream = confStream.broadcast(confDescriptor);
其中BroadcastStream和非BroadcastStream可以通过非广播流connect广播流来连接，返回的结果是一个BroadcastConnectedStream，非广播流可以是DataStream或者KeyedStream，我们可以调用process()进行数据的处理，具体的处理主要取决于非广播流是keyed还是non-keyed
- if that is keyed, then the function is a KeyedBroadcastProcessFunction.
- if it is non-keyed, the function is a BroadcastProcessFunction.
调用不同的处理类会有不同的实现方法
BroadcastProcessFunction和KeyedBroadcastProcessFunction这两个类如下所示：
代码语言：javascript复制public abstract class BroadcastProcessFunction<IN1, IN2, OUT> extends BaseBroadcastProcessFunction {
public abstract void processElement(IN1 value, ReadOnlyContext ctx, Collector<OUT> out) throws Exception;
public abstract void processBroadcastElement(IN2 value, Context ctx, Collector<OUT> out) throws Exception;
}public abstract class KeyedBroadcastProcessFunction<KS, IN1, IN2, OUT> {
public abstract void processElement(IN1 value, ReadOnlyContext ctx, Collector<OUT> out) throws Exception;
public abstract void processBroadcastElement(IN2 value, Context ctx, Collector<OUT> out) throws Exception;
public void onTimer(long timestamp, OnTimerContext ctx, Collector<OUT> out) throws Exception;
}- processBroadcastElement用于处理广播流中的数据，processElement用于处理主流中的数据。
- 两种方法在提供Context方面有所不，非广播方有ReadOnlyContext，而广播方有Context，也就是广播方具有写权限
- 两个Context 提供的功能
- 允许访问广播状态：ctx.getBroadcastState(MapStateDescriptor<K, V> stateDescriptor)
- 允许查询元素的时间戳：ctx.timestamp()
- 得到当前的水印：ctx.currentWatermark()
- 获得当前处理时间：ctx.currentProcessingTime()
- 将元素发射到侧输出：ctx.output(OutputTagoutputTag, X value)
- 此外keyed流多了一个onTimer方法，也就是他可以注册定时任务。
实例讲解
简述一下需求
- 统计每秒钟状态码非200的错误数和错误率
- 如果错误数大于指定的阈值则报警
- 阈值动态可配置
自定义source
首先通过自定义source和sql计算出来错误数和错误率
代码语言：javascript复制 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) )";
Table table = tenv.sqlQuery(sql);
DataStream<Result> dataStream = tenv.toAppendStream(table, Result.class);
然后我们自定义一个广播流的source，这里我们循环生成一些随机数来模拟生产环境，实际中我们可以去消费kafka的数据，或者循环去查询数据库。
代码语言：javascript复制
public static class BroadcastSource implements SourceFunction<Integer>{
@Override
public void run(SourceContext<Integer> ctx) throws Exception{
while (true){
Thread.sleep(3000);
ctx.collect(randInt(15, 20));
}
}
/**
* 生成指定范围内的随机数
* @param min
* @param max
* @return
*/
private int randInt(int min, int max){
Random rand = new Random();
int randomNum = rand.nextInt((max - min) 1) min;
return randomNum;
}
@Override
public void cancel(){
}
}
最后我们把两个流连接起来
代码语言：javascript复制
dataStream.connect(broadcastStream)
.process(new BroadcastProcessFunction<Result,Integer,Result>(){
@Override
public void processElement(
Result element,
ReadOnlyContext ctx,
Collector<Result> out) throws Exception{
Long v = ctx.getBroadcastState(confDescriptor)
.get("value");
if (v != null && element.getErrorcount() > v){
LOG.info("收到了一个大于阈值{}的结果{}.", v, element);
out.collect(element);
}
}
@Override
public void processBroadcastElement(
Integer value,
Context ctx,
Collector<Result> out) throws Exception{
ctx.getBroadcastState(confDescriptor)
.put("value", value.longValue());
}
});
生产环境下，我们可以把这个输出流通过process处理，发送到我们的报警系统。
完整的代码请参考
https://github.com/zhangjun0x01/bigdata-examples/blob/master/flink/src/main/java/cep/WebMonitorAlertDynamicConf.java
