flink使用06-如何处理窗口内的数据

上一节主要是大致介绍了下 flink 的窗口组成, 以及如何去划分窗口的. 那么这一篇文章主要是对剩下的内容做一下总结, 说一下如何对窗口内的数据做处理.

Window Function

Window Assigner 的作用是划分窗口的, 而 Window Function 就是对窗口内的数据做处理的一个过程. Flink 提供了 4 种类型的 Window Function, 分别是 ReduceFunction / AggregateFunction / FoldFunction / ProcessWindowFunction. 另外, 这四类还根据计算原理的不同分为增量聚合函数和全量窗口函数. 增量的计算性能比较高, 主要是基于中间状态的计算结果, 窗口中只维护中间结果的状态值.

1. ReduceFunction (增量)

对输入的两个相同类型的元素按照指定的计算方式进行聚合, 通过实现 ReduceFunction 接口就可以在reduce( ) 函数内部进行聚合操作了.

// 将Tuple2 按照 f1 进行 keyBy, 之后将 f0字符合并起来
input.keyBy(x -> x.f1)
    .timeWindow(Time.seconds(10), Time.seconds(1))
    .reduce(new ReduceFunction<Tuple2<String, Long>>() {
        @Override
        public Tuple2<String, Long> reduce(Tuple2<String, Long> t1, Tuple2<String, Long> t2) throws Exception {
            return new Tuple2<>(t1.f0 + t2.f0, t1.f1);
            }
        });

当然也可以使用匿名函数的方式,写起来会更加简洁.上述代码可以改为:

input.keyBy(x -> x.f1).timeWindow(Time.seconds(10), Time.seconds(1))
    .reduce((t1,t2) -> new Tuple2<>(t1.f0 + t2.f0, t1.f1));

2. AggregateFunction (增量)

AggregateFunction 相对于ReduceFunction更加灵活,但是实现起来也更复杂, AggregateFunction有 4 个需要复写的方法, 其中createAccumulator( ) 定义累加器, add( ) 定义数据的添加逻辑, getResult( ) 定义了根据 accumulator 计算结果的逻辑, merge()方法定义合并 accumulator 的逻辑.

input.keyBy(x -> x.f1)
    .timeWindow(Time.seconds(10), Time.seconds(1))
    // 自定义一个AggregateFunciton, 将相同标号 f1 的数据的 f0字符串字段合并在一起
    // ("hello", 1L) + ("world", 1L) = ("hello world", 1L)
    .aggregate(new MyAggregateFunction());

通过自定义的 MyAggregateFunction() 来实现 AggregateFunction 接口

public static class MyAggregateFunction implements AggregateFunction<Tuple2<String, Long>, String, String>{
        @Override
        public String createAccumulator() {
            // 初始化累加器
            return "";
        }
        @Override
        public String add(Tuple2<String, Long> t, String s) {
            // 输入数据与累加器的合并
            return s + " " +t.f0;
        }
        @Override
        public String getResult(String s) {
            // 得到累加器的结果
            return s.trim();
        }
        @Override
        public String merge(String s, String acc1) {
            // 合并累加器
            return s + " " + acc1;
        }
    }

3. FoldFunction (增量)

FoldFunction定义了如何将窗口中的输入元素与外部的元素合并的逻辑

input.keyBy(x -> x.f1)
.timeWindow(Time.seconds(10), Time.seconds(1)).fold("flink", (acc, t) ->t.f0 + acc);

FoldFunction在新版本已经被标记@Deprecated了, 建议使用AggregateFunction代替

4. ProcessWindowFunction (全量)

ProcessWindowFunction 相较于其他的 Window Function, 可以实现一些更复杂的计算, 比如基于整个窗口做某些指标计算 或者需要操作窗口中的状态数据和窗口元数据. Flink 提供了 ProcessWindowFunction 这个抽象类, 继承此类就可以实现ProcessWindowFunction, 其中, 必须要实现 process( ) 方法, 这是处理窗口数据的主要方法.还在一下跟窗口数据相关的方法可以有选择的实现.

public static class MyProcessWindowFunction extends ProcessWindowFunction<Tuple3<String, Long, Long>, String, Long, TimeWindow> {
    @Override
    public void process(Long s, Context context, Iterable<Tuple3<String, Long, Long>>       elements, Collector<String> out) throws Exception {
    // 统计每个窗口内的所有数据的 f0字段加起来共有多少个单词
    // 也就做单个窗口的 wordcount
    Long count = 0L;
    for (Tuple3<String, Long, Long> element : elements) {
    count += element.f0.split(" ").length;
    }
    out.collect("window: " + context.window() + " word count: " + count);
    }
}

5. 增量与全量共同使用

增量聚合函数虽然性能好, 但是灵活性不如全量函数, 例如对窗口状态数据的操作以及对窗口中的元数据信息的获取. 但是如果用 ProcessWindowFunction 去完成一些基础的增量计算相对比较浪费资源, 因此可以两者结合的方式来实现.

input.keyBy(x -> x.f1)
    .timeWindow(Time.seconds(10), Time.seconds(1))
    // 第一个Function为 ReduceFunction, 取窗口的最小值
    .reduce((r1, r2) -> {
    return r1.f0 < r2.f0 ? r1 : r2;
    // 第二个Function为 ProcessWindowFunction, 获取窗口的时间信息
    }, new ProcessWindowFunction<Tuple2<Long, Long>, String, Long, TimeWindow>() {
    @Override
    public void process(Long aLong, Context context, Iterable<Tuple2<Long, Long>>       elements, Collector<String> out) throws Exception {
    out.collect("window: " + context.window()); 
        }
    }).print();

Flink 窗口中的其他组件

除了 Window Assigner 和 Window Function外,Flink的窗口中还有 Triger窗口触发器, 其负责判断何时将窗口中的数据取出做计算, flink已经默认为各种类型的窗口实现了 triger. 用户也可以自己手动指定. Evictors 是数据剔除器, 目的是把窗口中的数据按照需求做一定的剔除. Flink也有 API 针对延迟数据做处理, 延迟的数据可以丢弃也可以通过sideOutputLateDate( ) 方法处理.

1. Triger 窗口触发器

EventTimeTrigger: 通过对比 watermark 和窗口 EndTime 确定是否触发窗口

ProcessTimeTrigger: 通过对比 ProcessTime 和窗口 EndTime 确定是否触发窗口

ContinuousEventTimeTrigger: 根据间隔时间周期性触发窗口

ContinuousEventTimeTrigger: 同上, 区别是使用ProcessTime

CountTrigger: 根据接入数量是否超过阈值

DeltaTrigger: 根据计算出来的 Delta 指标是否超过指定的 Threshold

PurgingTrigger: 可以将任意触发器作为参数转换为Purge类型触发器

2. Evictors触发器

CountEvictor: 保持固定数量的数据, 超过的剔除

DeltaEvictor: 通过定义 delta 和 threshold , 计算两个数据之间的 delta 值, 超过则剔除

TimeEvictor: 指定时间间隔, 将当前窗口中的最新元素的时间减去Interval, 然后将小于该结果的数据全部剔除