基于Flink的实时告警实现(3)：时间窗口

本专题将会从0到1实现告警处理流程，并会讲解实现过程中使用到的Flink中的技术。

1 告警需求

前面的(1)和(2)简单的介绍了flink、开发环境的搭建以及读取kafka的数据并对数据进行转换以便后续处理，现在就可以开始真正的实现告警了。先简单说下需求：

该需求的主要目的是对网络交换机进行监控，快速发现网络的问题。采集器接收到网络交换机的指标数据，例如错包等，将这些指标存入到kafka中，然后再从kafka中读取数据并写入到后端的持久化存储。这里需要添加告警功能，常规的方法可能是另外写一个程序，通过后端的存储(例如TSDB、clickhouse)的查询语句做告警判断，例如sql等，然后根据给定的告警规则(例如，流量占比>80%)进行告警。这种方式存在的问题是：后端的存储和查询是有延迟的，特别是分布式环境下，数据需要同步，那么再从里面去查询进行告警的延迟也比较大。为了解决这个问题，可以利用flink的流式计算减少告警延迟。

在网络监控层面常用的两个指标有：流量使用占比(当前流量/接口带宽)，错包数(一段时间产生的错包数)。通常，从交换机采集到的数据都是计数器，例如，接口流入的总的字节数，通过该指标可以计算出当前流量(某段时间内流过的字节数/某段时间)。因此，为了监控流量和错包，需要对一个时间段的数据进行计算，而且这个时间段应该尽可能短。当前数据采集的粒度是30秒一个点，因此，把这个时间段定为2分钟，通过这2分钟里面的4个点进行计算和判断。

从告警的整体流程上来说，告警模块需要使用两份数据：指标数据和告警规则。从kafka读取指标数据，然后与告警规则进行对比，如果告警规则满足条件，则推送告警。因此，整个告警模块的流程如下：

读取kafka的数据，然后解析成对象数据
获取近2分钟的数据，通过2分钟的4个点计算出流量占比和错包数
将流量占比和错包数与告警规则对比，如果满足条件，则发送告警

2 Flink中的时间

Flink中的时间有3种：Event Time(事件产生的时间，通常存在于数据中)，Ingestion Time(进入Flink的时间)，Processing Time(进入算子的时间)。那么，我们在计算流量时应该用哪个时间呢？数据本身里面就包含时间戳字段，因此，在计算流量时，肯定也是用数据里面的时间戳。例如，第一条数据的时间戳是t1，采集的数据是v1，最后一条数据的时间戳是t2，采集的数据是v2，那么，流量就是(v2-v1)/(t2-t1)。

3 Flink中的时间窗口

Flink中的时间窗口分为三种类型：滚动时间窗口、滑动时间窗口、会话窗口。

滚动时间窗口：每隔一段时间滚动一次，窗口和窗口之间不重合，因此，创建窗口时只有一个参数(窗口大小)，例如，窗口大小为2min，当12:00:00_{12:02:00为第一个窗口时，12:02:00}12:04:00为第二个窗口。

stream.keyBy(0).timeWindow(Time.minutes(2)).apply();

滑动时间窗口：每隔一段时间滑动一次，窗口和窗口之间通常是重合的，因此，创建窗口时有两个参数(滑动步长和窗口大小)，例如，滑动步长为30s，窗口大小为2min，当12:00:00_{12:02:00为第一个窗口时，12:00:30}12:02:30为第二个窗口。

stream.keyBy(0).timeWindow(Time.minutes(2), Time.seconds(30)).apply();

4 基于Flink的时间实现告警

根据前面的需求和Flink中对时间窗口的支持，可以用以下方式实现告警：

将从交换机采集到的数据解析成对象，当前包含5个属性：指标名、交换机、端口、时间、指标值。然后按照交换机和端口进行分区，再使用滑动窗口计算流量或者错包就得到当前时刻的指标值，将计算得到的指标值与告警策略中的指标值对比就可以判断是否应该推送告警。

kafkaStream
            .keyBy("metric", "device", "intf")
            .timeWindow(Time.minutes(2), Time.Seconds(30))
            .apply(new RichWindowFunction<SwitchMetric, Object, Tuple, TimeWindow>() {
                @Override
                public void apply(Tuple tuple, TimeWindow timeWindow, Iterable<SwitchMetric> iterable, Collector<object> collector) {
                    // Todo: 此处就可以编写业务逻辑：
                    // 从iterable中读取窗口中的数据，计算出指标值，然后与规则进行对比
                }
            })

4 小结

本文对告警的逻辑进行了简单讲解，然后通过使用Flink中的滑动时间窗口计算指标值，再根据计算得到的指标值与告警规则对比，如果规则判断成功，就可以推送告警。