简介
使用方法
简述一下需求
自定义source

背景

紧接着上一篇文章【FLINK实战-使用CEP进行网站监控报警和报警恢复】，上一篇主要讲了怎么对数据流做监控报警，但是实际生产环境中，可能我们对这个报警的阈值设置成多大也没有一个准，可能需要根据经验不断的来修改，所以就涉及了可能需要不断的修改这个报警的阈值，但是如果每次修改了之后，都通过重启flink程序来实现，这个成本就有点高了，所以我们这次主要是讲解一下，如何使用flink的广播动态的更新配置来设置这个报警的阈值.

flink broadstate

简介

是flink提供的一种算子，可以使用一个Stream接收不断变化的数据（比如我们的配置数据），然后把这些数据广播到flink的所有task中，这样主Stream中的数据就能动态的从广播流中获取所需要的配置，然后根据动态的配置来处理数据.

使用方法

构造广播流


//构造一个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这两个类如下所示：

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 提供的功能
1. 允许访问广播状态：ctx.getBroadcastState(MapStateDescriptor<K, V> stateDescriptor)
2. 允许查询元素的时间戳：ctx.timestamp()
3. 得到当前的水印：ctx.currentWatermark()
4. 获得当前处理时间：ctx.currentProcessingTime()
5. 将元素发射到侧输出：ctx.output(OutputTagoutputTag, X value)
此外keyed流多了一个onTimer方法，也就是他可以注册定时任务。

实例讲解

简述一下需求

统计每秒钟状态码非200的错误数和错误率
如果错误数大于指定的阈值则报警
阈值动态可配置

自定义source

首先通过自定义source和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) )";

		Table table = tenv.sqlQuery(sql);
		DataStream<Result> dataStream = tenv.toAppendStream(table, Result.class);

然后我们自定义一个广播流的source，这里我们循环生成一些随机数来模拟生产环境，实际中我们可以去消费kafka的数据，或者循环去查询数据库。


	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(){

		}
	}

最后我们把两个流连接起来


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

flink实战-使用广播实现报警阈值动态更新

背景

flink broadstate

简介

使用方法

实例讲解

简述一下需求

自定义source