前往小程序,Get更优阅读体验!
立即前往
首页
学习
活动
专区
工具
TVP
发布
社区首页 >专栏 >flink实战-使用广播实现报警阈值动态更新

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

作者头像
大数据技术与应用实战
发布2020-09-15 14:29:51
1.5K0
发布2020-09-15 14:29:51
举报
文章被收录于专栏:大数据技术与应用实战
  • 简介
  • 使用方法
  • 简述一下需求
  • 自定义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;
}
代码语言:javascript
复制
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方法,也就是他可以注册定时任务。

实例讲解

简述一下需求

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

自定义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

本文参与 腾讯云自媒体同步曝光计划,分享自微信公众号。
原始发表:2020-06-05,如有侵权请联系 cloudcommunity@tencent.com 删除

本文分享自 大数据技术与应用实战 微信公众号,前往查看

如有侵权,请联系 cloudcommunity@tencent.com 删除。

本文参与 腾讯云自媒体同步曝光计划  ,欢迎热爱写作的你一起参与!

评论
登录后参与评论
0 条评论
热度
最新
推荐阅读
目录
  • 背景
  • flink broadstate
    • 简介
      • 使用方法
      • 实例讲解
        • 简述一下需求
          • 自定义source
          相关产品与服务
          大数据
          全栈大数据产品,面向海量数据场景,帮助您 “智理无数,心中有数”!
          领券
          问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档