前言
祝广大女性节日快乐~
快问快答
- Flink DataStream API中的RichFunction有哪些用途/特点？
- RichFunction中获取到的RuntimeContext是干什么用的？
- 所有Function都有对应的RichFunction实现吗？
- 所有Flink流处理的算子都可以传入RichFunction吗？
前两个问题实际上可以合并成一个问题。RichFunction的特点是比Function多出了生命周期管理（open()和close()方法），以及能够获取其运行时上下文RuntimeContext。RuntimeContext与Function的每个并行实例（即一个Sub-task）相关联，通过它还能进一步得到如下信息：
- 运行时静态信息，如Task的名称、并行度、最大并行度、当前Sub-task的编号、当前类加载器等；
- 全局数据结构，即累加器（Accumulators）、广播变量（Broadcast variables）和分布式缓存（Distributed cache）；
- 创建各种状态句柄，即我们熟知的
get***State(StateDescriptor)方法。
第三个问题，yes；第四个问题，no。
RichFunction不适用的场景
简单的开窗聚合场景：
dataStream.keyBy(x -> x.getKey()) .window(TumblingProcessingTimeWindows.of(Time.seconds(1))) .reduce(new MyRichReduceFunction<>()) 这段代码能编译通过，但执行时会抛出UnsupportedOperationException，提示ReduceFunction of reduce can not be a RichFunction。如果换成aggregate()方法和RichAggregateFunction会有同样的问题，提示This aggregation function cannot be a RichFunction。在WindowedStream的对应实现中，可以看到此路不通：
public SingleOutputStreamOperator<T> reduce(ReduceFunction<T> function) { if (function instanceof RichFunction) { throw new UnsupportedOperationException( "ReduceFunction of reduce can not be a RichFunction. " + "Please use reduce(ReduceFunction, WindowFunction) instead."); } // clean the closure function = input.getExecutionEnvironment().clean(function); return reduce(function, new PassThroughWindowFunction<>()); } public <ACC, R> SingleOutputStreamOperator<R> aggregate(AggregateFunction<T, ACC, R> function) { checkNotNull(function, "function"); if (function instanceof RichFunction) { throw new UnsupportedOperationException( "This aggregation function cannot be a RichFunction."); } TypeInformation<ACC> accumulatorType = TypeExtractor.getAggregateFunctionAccumulatorType( function, input.getType(), null, false); TypeInformation<R> resultType = TypeExtractor.getAggregateFunctionReturnType( function, input.getType(), null, false); return aggregate(function, accumulatorType, resultType); } 为什么不能用Rich[Reduce / Aggregate]Function？
答案并不难：与FlatMap、Filter等算子不同，Reduce和Aggregate本身就是自带确定的状态语义的算子，不需要用户手动操作状态（如果用户能干预的话大概率会出问题），也不需要生命期管理的特性（它们的生命期总是始于第一条数据，终于最后一条数据）。
以Reduce逻辑为例（Aggregate同理），不妨进一步看下对应的窗口算子是如何构造的。
public <R> WindowOperator<K, T, ?, R, W> reduce( ReduceFunction<T> reduceFunction, WindowFunction<T, R, K, W> function) { Preconditions.checkNotNull(reduceFunction, "ReduceFunction cannot be null"); Preconditions.checkNotNull(function, "WindowFunction cannot be null"); if (reduceFunction instanceof RichFunction) { throw new UnsupportedOperationException( "ReduceFunction of apply can not be a RichFunction."); } if (evictor != null) { return buildEvictingWindowOperator( new InternalIterableWindowFunction<>( new ReduceApplyWindowFunction<>(reduceFunction, function))); } else { ReducingStateDescriptor<T> stateDesc = new ReducingStateDescriptor<>( WINDOW_STATE_NAME, reduceFunction, inputType.createSerializer(config)); return buildWindowOperator( stateDesc, new InternalSingleValueWindowFunction<>(function)); } } 注意到这里创建了ReducingStateDescriptor（ReduceFunction恰好是它的一个入参），并最终获取了内置的ReducingState句柄。其实就DataStream API用户的日常编程习惯而言，很少会主动用到ReducingState（以及AggregateState）。即使这样，在它们的描述符构造方法中，也加了同样的强制校验，防止传入RichFunction，以保护状态的确定性。
public ReducingStateDescriptor( String name, ReduceFunction<T> reduceFunction, Class<T> typeClass) { super(name, typeClass, null); this.reduceFunction = checkNotNull(reduceFunction); if (reduceFunction instanceof RichFunction) { throw new UnsupportedOperationException( "ReduceFunction of ReducingState can not be a RichFunction."); } } 话说回来，Rich[Reduce / Aggregate]Function在Flink工程内部以及示例中都没有有效的使用过，所以我们大概可以判定这是Flink发展过程中的遗产吧（笑
The End
晚安晚安。