09-Flink 处理函数之广播流-动态更新阈值

Flink 官网主页地址：https://flink.apache.org
Flink 官方中文地址：https://nightlies.apache.org/flink/flink-docs-stable/zh/

背景

在实际生产环境中，对报警的阈值需要根据经验不断的来修改，但是如果每次修改了之后，都通过重启 Flink 程序来实现，这个成本就有点高了，所以我们这次主要是讲解一下，如何使用 Flink 的广播动态的更新配置来设置这个报警的阈值。

flink broadstate

Flink 中的广播流（Broadcast Stream）是一种特殊的数据流，它允许将一个流的数据广播到其他所有的流上。这样主 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

keyed 键控流对应：KeyedBroadcastProcessFunction.
非键控流对应：BroadcastProcessFunction.

调用不同的处理类会有不同的实现方法

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(OutputTag outputTag, 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());
            }
        });