Flink窗口状态大小和状态管理

Question

在阅读了Flink的文档并四处搜索后，我无法完全理解Flink的句柄在其窗口中的状态。假设我有一个每小时滚动的窗口，其中包含一个聚合函数，该函数将消息累积到某个java pojo或scala case类中。该窗口的大小将与一小时内进入该窗口的事件数量相关联，还是仅仅与POJO/Case类相关联，因为我将事件累加到该对象中。（例如，如果将10000个味精数成一个整数，大小会接近10000*味精大小还是int大小？）另外，如果im使用POJO或case类，flink是否为我处理状态（如果内存耗尽，则溢出到磁盘/在检查点保存状态等），或者我必须为此使用flink的状态对象？

谢谢你的帮助！

Answer 1

窗口的状态大小取决于应用的函数类型。如果应用reducefunction或aggregatefunction，到达的数据将立即聚合，窗口只保存聚合的值。如果应用processWindowfunction或windowfunction，Flink将收集所有输入记录，并在时间（取决于窗口类型的事件或处理时间）超过窗口结束时间时应用该函数。

您还可以将这两种类型的函数组合在一起，即具有AggregateFunction，后跟ProcessWindowFunction。在这种情况下，到达的记录会立即聚合，当窗口关闭时，聚合结果会作为单个值传递给processWindowfunction。这很有用，因为您可以进行增量聚合（由于reducefunction/aggregatefunction)，还可以访问窗口元数据，如开始和结束时间戳（由于processWindowfunction)。

如何管理状态取决于所选择的状态后端。如果配置fsstateBackend，则所有本地状态都保留在TaskManager的堆中，如果状态变得太大，则使用OutofMemoryError杀死JVM进程。如果配置RockSDBStateBackend，则状态会溢出到磁盘。每一次状态访问都需要de/serialization成本，但为状态提供了更多的存储空间。