本文全部来源于官方文档解释。
Window Apply?#
WindowedStream → DataStream?#
AllWindowedStream → DataStream?#
Applies a general function to the window as a whole. Below is a function that manually sums the elements of a window.
解释：将一个通用的函数应用到整个窗口。可见apply用于窗口函数之后。
举例：
windowedStream.apply(new WindowFunction<Tuple2<String,Integer>, Integer, Tuple, Window>() {
public void apply (Tuple tuple,
Window window,
Iterable<Tuple2<String, Integer>> values,
Collector<Integer> out) throws Exception {
int sum = 0;
for (value t: values) {
sum += t.f1;
}
out.collect (new Integer(sum));
}
});
// applying an AllWindowFunction on non-keyed window stream
allWindowedStream.apply (new AllWindowFunction<Tuple2<String,Integer>, Integer, Window>() {
public void apply (Window window,
Iterable<Tuple2<String, Integer>> values,
Collector<Integer> out) throws Exception {
int sum = 0;
for (value t: values) {
sum += t.f1;
}
out.collect (new Integer(sum));
}
});源码的角度分析：
接口函数windowFuntion中 定义了apply方法。
源码的注释为：计算窗口的值，并输出none或多个元素。 ?
总结：apply 是一个通用的函数应用于窗口中数据的计算，可以紧挨window? Funtion之后使用，另外在接口方法 WindowFunction中定义了apply方法，用户可以自定义apply对窗口中数据的处理规则。
Process Function?#
The ProcessFunction?#
ProcessFunction是一种底层流处理操作，可以访问所有（非循环）流应用程序的基本构建块：
- 事件（流元素）
- 状态（容错，一致，仅在键控流上）
- 计时器（事件时间和处理时间，仅在键控流上）
可以将其ProcessFunction视为可以FlatMapFunction访问键控状态和计时器。它通过为输入流中接收到的每个事件调用来处理事件。
对于容错状态，ProcessFunction可以访问 Flink 的keyed state，可以通过 访问?RuntimeContext，类似于其他有状态函数访问 keyed state 的方式。
计时器允许应用程序对处理时间和事件时间的变化做出反应。对该函数的每次调用processElement(...)都会获得一个Context对象，该对象可以访问元素的事件时间时间戳和TimerService。可TimerService用于为将来的事件/处理时间瞬间注册回调。对于事件时间计时器，onTimer(...)当当前水印超过或超过计时器的时间戳时调用该方法，而对于处理时间计时器，onTimer(...)当挂钟时间达到指定时间时调用该方法。在该调用期间，所有状态再次限定为创建计时器的键，允许计时器操作键控状态。
如果要访问键控状态和计时器，则必须
ProcessFunction在键控流上应用：
https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/dev/datastream/operators/process_function/?总结：processFunction 是一个底层的处理流的函数：它能为用户提供三方面东西，1.元素2.状态3.计时器 可以让用户很方便的定义流数据的规则。一般我们通常使用ProcessFunction较多。
ReduceFunction?#
ReduceFunction指定输入中的两个元素如何组合生成相同类型的输出元素。?Flink使用ReduceFunction递增地聚合窗口的元素。 ?
val input: DataStream[(String, Long)] = ...
input
.keyBy(<key selector>)
.window(<window assigner>)
.reduce { (v1, v2) => (v1._1, v1._2 + v2._2) }AggregateFunction
AggregateFunction是ReduceFunction的一般化版本，
它有三种类型:
输入类型(IN)、
累加类型(ACC)
输出类型(OUT)。
?输入类型是输入流中元素的类型，AggregateFunction有一个将一个输入元素添加到累加器的方法。?该接口还提供了创建初始累加器、将两个累加器合并到一个累加器以及从累加器提取输出(类型为OUT)的方法。?我们将在下面的例子中看到它是如何工作的。 输入类型和输出类型可以不相同?
与ReduceFunction一样，Flink会在一个窗口的输入元素到达时递增地聚合它们。 ?
/**
* The accumulator is used to keep a running sum and a count. The [getResult] method
* computes the average.
*/
class AverageAggregate extends AggregateFunction[(String, Long), (Long, Long), Double] {
override def createAccumulator() = (0L, 0L)
override def add(value: (String, Long), accumulator: (Long, Long)) =
(accumulator._1 + value._2, accumulator._2 + 1L)
override def getResult(accumulator: (Long, Long)) = accumulator._1 / accumulator._2
override def merge(a: (Long, Long), b: (Long, Long)) =
(a._1 + b._1, a._2 + b._2)
}
val input: DataStream[(String, Long)] = ...
input
.keyBy(<key selector>)
.window(<window assigner>)
.aggregate(new AverageAggregate)?
?
?
?
?
?
?
?