为什么window的reduceFunction不支持RichFunction呢

[nobleyd]

问题如title，当然当前Flink的API还是可以实现RichReduceFunction的效果的，就是基于windowFunction或者processWindowFuntion。
但是，windowFunc和reduceFunc最大的区别在于windowFunc是窗口触发操作，而reduce是实时的增量操作。
如果说我的窗口计算对于每个record的到来都需要一个极复杂的操作，我更希望在reduce中完成，而不是windowFunc中完成，因为那样会导致整个窗口所有key的数据几乎在同一时刻触发，这回导致压力变高。

[cs]

flink不是有现成的RichReduceFunction吗




------------------ 原始邮件 ------------------
发件人:                                                                                                                        "user-zh"                                                                                    <[hidden email]&gt;;
发送时间:&nbsp;2021年1月28日(星期四) 晚上8:26
收件人:&nbsp;"user-zh"<[hidden email]&gt;;

主题:&nbsp;为什么window的reduceFunction不支持RichFunction呢



问题如title，当然当前Flink的API还是可以实现RichReduceFunction的效果的，就是基于windowFunction或者processWindowFuntion。
但是，windowFunc和reduceFunc最大的区别在于windowFunc是窗口触发操作，而reduce是实时的增量操作。
如果说我的窗口计算对于每个record的到来都需要一个极复杂的操作，我更希望在reduce中完成，而不是windowFunc中完成，因为那样会导致整个窗口所有key的数据几乎在同一时刻触发，这回导致压力变高。

[Smile]

Hi, nobleyd,

请问你是在哪个版本发现 reduceFunction 不支持 RichFunction 呢？
我在1.12 版本试了一下是支持的呀，而且 JavaDoc 里也有 RichReduceFunction 类[1]

[1].
https://ci.apache.org/projects/flink/flink-docs-release-1.12/api/java/org/apache/flink/api/common/functions/RichReduceFunction.html





--
Sent from: http://apache-flink.147419.n8.nabble.com/

[Kezhu Wang]

reduceFunction 是和 ReducingStateDescriptor 配合的， 并不是
“window的”。“WindowOperator” 的 function 是 “InternalWindowFunction”，这个可以是
“RichFunction”。

Flink 中的 state 除了可以绑定 key，也可以绑定 namespace，只是 namespace
并没有直接暴露给用户。如果需要的话，可以自己写个 WindowOperator，暴露个 WindowFunction。

Interface WindowFunction {
    // You could do incremental aggregation here.
    void processElement(Context context, Window window, Element element);

    void fireWindow(Context context, Window window);
}

interface WindowedRuntimeContext {
     State getWindowedState(StateDescriptor descriptor).
}

把所有 window concepts 交给 WindowOperator，WindowFunction 作具体业务。

On January 28, 2021 at 20:26:47, 赵一旦 ([hidden email]) wrote:

问题如title，当然当前Flink的API还是可以实现RichReduceFunction的效果的，就是基于windowFunction或者processWindowFuntion。

但是，windowFunc和reduceFunc最大的区别在于windowFunc是窗口触发操作，而reduce是实时的增量操作。
如果说我的窗口计算对于每个record的到来都需要一个极复杂的操作，我更希望在reduce中完成，而不是windowFunc中完成，因为那样会导致整个窗口所有key的数据几乎在同一时刻触发，这回导致压力变高。

[nobleyd]

不是，flink是提供了richReduce，但不支持基于window的richReduce。
基于window的reduce只支持简单reduce，具体需要做自定义状态计算的，按照注释要求使用windowFunction和ProcessWindowFunction去做呀。

一直都是这样，1.12也是的哈。

Kezhu Wang <[hidden email]> 于2021年1月29日周五 上午11:40写道：

> reduceFunction 是和 ReducingStateDescriptor 配合的， 并不是
> “window的”。“WindowOperator” 的 function 是 “InternalWindowFunction”，这个可以是
> “RichFunction”。
>
> Flink 中的 state 除了可以绑定 key，也可以绑定 namespace，只是 namespace
> 并没有直接暴露给用户。如果需要的话，可以自己写个 WindowOperator，暴露个 WindowFunction。
>
> Interface WindowFunction {
>     // You could do incremental aggregation here.
>     void processElement(Context context, Window window, Element element);
>
>     void fireWindow(Context context, Window window);
> }
>
> interface WindowedRuntimeContext {
>      State getWindowedState(StateDescriptor descriptor).
> }
>
> 把所有 window concepts 交给 WindowOperator，WindowFunction 作具体业务。
>
> On January 28, 2021 at 20:26:47, 赵一旦 ([hidden email]) wrote:
>
> 问题如title，当然当前Flink的API还是可以实现RichReduceFunction的效果的，就是基于windowFunction或者processWindowFuntion。
>
> 但是，windowFunc和reduceFunc最大的区别在于windowFunc是窗口触发操作，而reduce是实时的增量操作。
> 如果说我的窗口计算对于每个record的到来都需要一个极复杂的操作，我更希望在reduce中完成，而不是windowFunc中完成，因为那样会导致整个窗口所有key的数据几乎在同一时刻触发，这回导致压力变高。
>
>
>

[nobleyd]

@Kezhu Wang <[hidden email]>
 你的说法大概懂。不过改造成本有点高，本身复写个算子不难。但是貌似这个地方考虑点太多了。比如我放个接口允许reduce的时候支持一个RichReduceFunction2。相当于对于每个record，要经过默认的reduceFunc，再经过我的RichReduceFunction2，然后窗口触发再去windowFunc。
这样可能我还需要调整emitWindowContent那部分，想办法将RichReduceFunction2中自定义的状态内容也丢进
windowFunc的参数中去。

赵一旦 <[hidden email]> 于2021年1月29日周五 下午12:34写道：

> 不是，flink是提供了richReduce，但不支持基于window的richReduce。
>
> 基于window的reduce只支持简单reduce，具体需要做自定义状态计算的，按照注释要求使用windowFunction和ProcessWindowFunction去做呀。
>
> 一直都是这样，1.12也是的哈。
>
> Kezhu Wang <[hidden email]> 于2021年1月29日周五 上午11:40写道：
>
>> reduceFunction 是和 ReducingStateDescriptor 配合的， 并不是
>> “window的”。“WindowOperator” 的 function 是 “InternalWindowFunction”，这个可以是
>> “RichFunction”。
>>
>> Flink 中的 state 除了可以绑定 key，也可以绑定 namespace，只是 namespace
>> 并没有直接暴露给用户。如果需要的话，可以自己写个 WindowOperator，暴露个 WindowFunction。
>>
>> Interface WindowFunction {
>>     // You could do incremental aggregation here.
>>     void processElement(Context context, Window window, Element element);
>>
>>     void fireWindow(Context context, Window window);
>> }
>>
>> interface WindowedRuntimeContext {
>>      State getWindowedState(StateDescriptor descriptor).
>> }
>>
>> 把所有 window concepts 交给 WindowOperator，WindowFunction 作具体业务。
>>
>> On January 28, 2021 at 20:26:47, 赵一旦 ([hidden email]) wrote:
>>
>> 问题如title，当然当前Flink的API还是可以实现RichReduceFunction的效果的，就是基于windowFunction或者processWindowFuntion。
>>
>> 但是，windowFunc和reduceFunc最大的区别在于windowFunc是窗口触发操作，而reduce是实时的增量操作。
>> 如果说我的窗口计算对于每个record的到来都需要一个极复杂的操作，我更希望在reduce中完成，而不是windowFunc中完成，因为那样会导致整个窗口所有key的数据几乎在同一时刻触发，这回导致压力变高。
>>
>>
>>

[likai]

你好， 请问想用reduce 里面需要存储状态，可以用 aggregate() 吗？ 看了一下代码，窗口的话，是把数据存储在窗口里面的状态。reduce 会生成 reduce 对应的状态存储放入窗口。任务窗口函数是不是自定义的，里面的状态也不能自定义。有聚合的情况下，只是把聚合函数作用在了窗口的状态上，里面的状态是要保存聚合结果的状态。 可以任务 窗口加聚合是一个算子。


likai
[hidden email]



> 在 2021年1月29日，下午12:49，赵一旦 <[hidden email]> 写道：
>
> windowFunc