Apache Flink源码解析之stream-windowfunction

本是古典何须时尚 2022-01-30 04:21 280阅读 0赞

`Window`也即窗口，是Flink流处理的特性之一。前一篇文章我们谈到了`Winodw`的相关概念及其实现。窗口的目的是将无界的流转换为有界的元素集合，但这还不是最终的目的，最终的目的是在这有限的集合上**apply**(应用)某种函数，这就是我们本篇要谈的主题——`WindowFunction`（窗口函数）。

那么窗口函数会在什么时候被应用呢？还记得上篇文章我们谈到了触发器`Trigger`，在触发器触发后会返回`TriggerResult`这个枚举类型的其中一个枚举值。当返回的是`FIRE`或者`FIRE_AND_PURGE`时，窗口函数就会在窗口上应用。

Flink中将窗口函数分为两种：

*  AllWindowFunction : 针对全局的不基于某个key进行分组的window的窗口函数的实现
 *  WindowFunction : 针对基于某个key进行分组的window的窗口函数的实现

它们在类型继承体系中分属两个不同的体系：

![flink-stream-window-function\_all-class-diagram][flink-stream-window-function_all-class-diagram]

但可以看到，针对这两个体系几乎都提供了相同功能的窗口函数的实现。

# AllWindowFunction #

所有不基于某个key进行分组的window的窗口函数的实现的基类。该接口是个泛型接口，需要指定三个泛型参数：

*  IN ：input数据的类型
 *  OUT ：output对象的类型
 *  W : 继承自`Window`，表示需要在其上应用该操作的`Window`的类型

该接口只有一个接口方法：

void apply(W window, Iterable<IN> values, Collector<OUT> out) throws Exception;

该方法用于在`window`上的元素集合`values`进行计算，然后`out`出0个或多个值。

## RichAllWindowFunction ##

抽象类，继承`AbstractRichFunction`以提供**rich** 的AllWindowFunction（`AbstractRichFunction`提供了open/close方法对以及获得运行时上下文对象的手段）。我们在之前解析`SourceFunction`和`SinkFunction`时多次看到这种实现模式。这里该类不提供任何实现。

## ReduceIterableAllWindowFunction ##

`ReduceIterableAllWindowFunction`用于对其窗口内的所有元素迭代应用`reduce`操作并合并为一个元素，然后再发射出去。它接收`ReduceFunction`的实例，以提供reduce函数。

该类apply方法实现如下：

public void apply(W window, Iterable<T> input, Collector<T> out) throws Exception {
    
            T curr = null;
            for (T val: input) {
                if (curr == null) {
                    curr = val;
                } else {
                    curr = reduceFunction.reduce(curr, val);
                }
            }
            out.collect(curr);
        }

`reduceFunction#reduce`方法，用于将第一个参数和第二个参数进行合并为一个元素。

## ReduceApplyAllWindowFunction ##

`ReduceApplyAllWindowFunction`用于对窗口内的所有元素进行**reduce**操作后再进行调用**apply**。其构造器接收两个参数：

*  reduceFunction ： 提供reduce操作的`ReduceFunction`
 *  windowFunction ： 提供apply操作的`AllWindowFunction`，该参数用于对window中元素进行reduce之后产生的单个元素再进行最终的apply操作。

该类的apply实现如下：

public void apply(W window, Iterable<T> input, Collector<R> out) throws Exception {
    
            T curr = null;
            for (T val: input) {
                if (curr == null) {
                    curr = val;
                } else {
                    curr = reduceFunction.reduce(curr, val);
                }
            }
            windowFunction.apply(window, Collections.singletonList(curr), out);
        }

## PassThroughAllWindowFunction ##

`PassThroughAllWindowFunction`该类仅仅提供***passthrough***功能，也即直接通过发射器将窗口内的元素迭代发射出去，除此之外不进行任何操作。

## FoldApplyAllWindowFunction ##

`FoldApplyAllWindowFunction`用于对窗口中的数据先进行**fold**操作，得到一个最终合并的元素，再进行**apply**操作。因此它需要如下三个参数：

*  initialValue ： 应用`foldFunction`的初始值
 *  foldFunction ：执行fold操作
 *  windowFunction ：对fold之后的最终值应用apply操作

该类继承自`WrappingFunction`。`WrappingFunction`类似于一个包装器，包装传进来的某个`Function`，给一些模式化方法(open/close)提供了一些便捷处理。

这里有一点需要区分一下，因为`ReduceFunction`和`FoldFuction`都具有将一组元素合并为单个元素的功能，所以他们看起来非常相似。不过他们还是有区别的，其中的一个区别就是，`FoldFunction`在进行fold操作的时候，还会进行潜在的类型转换。看下面的示例：

ReduceFunction<Integer> {
    
      public Integer reduce(Integer a, Integer b) { return a + b; }
    }
    
    [1, 2, 3, 4, 5] -> reduce()  means: ((((1 + 2) + 3) + 4) + 5) = 15

FoldFunction<String, Integer> {
    
      public String fold(String current, Integer i) { return current +
    String.valueOf(i); }
    }
    
    [1, 2, 3, 4, 5] -> fold("start-")  means: ((((("start-" + 1) + 2) + 3) + 4)
    + 5) = "start-12345" (as a String)

# WindowFunction #

这是Flink的另一个基于key进行分组的`WindowFunction`。因此跟`AllWindowFunction`主要的不同的是，其泛型参数多了一个`KEY`，表示进行分组的key的类型。

同时其接口方法中也相应多了一个参数：

void apply(KEY key, W window, Iterable<IN> input, Collector<OUT> out) throws Exception;

> 具体的实现跟`AllWindowFunction`的实现大同小异，不再多谈。

# 小结 #

本篇主要剖析了Flink提供的两种不同的窗口函数`AllWindowFunction`以及`WindowFunction`。并对Flink针对`AllWindowFunction`的实现进行了解读。

原文发布时间为：2016-05-12

本文作者：vinoYang

本文来自云栖社区合作伙伴[CSDN博客][CSDN]，了解相关信息可以关注CSDN博客。

[flink-stream-window-function_all-class-diagram]: http://7xkaaz.com1.z0.glb.clouddn.com/flink-stream-window-function_all-class-diagram.png
[CSDN]: http://blog.csdn.net/yanghua_kobe/article/details/51388359