说明

几乎所有Flink 函数类都有其Rich 版本

例如：

MapFunction 的富函数：RichMapFunction
FlatMapFunction 的富函数：RichFlatMapFunction
FilterFunction 的富函数：RichFilterFunction
ReduceFunction 的富函数：RichReduceFunction

应用场景

需要在函数处理之前，连接数据库（只连接一次），执行结束之后还要关闭数据库
获取得运行时相关的上下文信息

功能

可以获取 运行环境的上下文，并拥有一些生命周期方法，所以可以实现更复杂的功能

生命周期方法有：

open() 方法：初始化方法，当算子（如 map 或者filter）被调用之前， open() 会被调用，只会执行一次。如：连接数据库
close() 方法：是生命周期中的 最后一个调用的方法，做一些清理工作。如：关闭数据库

getRuntimeContext()

父类的 getRuntimeContext() 方法：提供了 RuntimeContext 的一些信息，例如：

函数执行的并行度
任务的名字
以及state 状态

案例

package filter;

import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class TestFilter2 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);

        DataStreamSource<Long> dss = env.fromSequence(1, 10);

        SingleOutputStreamOperator<Long> filter = dss.map(new MyMap());

        filter.print();

        env.execute();
    }


    public static class MyMap extends RichMapFunction<Long, Long> {
        @Override
        public void open(Configuration parameters) throws Exception {
            String taskName = getRuntimeContext().getTaskName();
            int index = getRuntimeContext().getIndexOfThisSubtask();
            System.out.println("open()-----subtask:"+index+"--taskName:["+taskName+"]");
        }

        @Override
        public void close() throws Exception {
            String taskName = getRuntimeContext().getTaskName();
            int index = getRuntimeContext().getIndexOfThisSubtask();
            System.out.println("close()-----subtask:"+index+"--taskName["+taskName+"]");
        }

        @Override
        public Long map(Long data) throws Exception {
            String taskName = getRuntimeContext().getTaskName();
            int index = getRuntimeContext().getIndexOfThisSubtask();
            System.out.println("map()-----subtask:"+index+"--taskName["+taskName+"]");

            return data;
        }
    }
}

执行结果：

open()-----subtask:1--taskName:[Source: Sequence Source -> Map -> Sink: Print to Std. Out]
open()-----subtask:0--taskName:[Source: Sequence Source -> Map -> Sink: Print to Std. Out]
map()-----subtask:1--taskName[Source: Sequence Source -> Map -> Sink: Print to Std. Out]
2> 1
map()-----subtask:1--taskName[Source: Sequence Source -> Map -> Sink: Print to Std. Out]
map()-----subtask:0--taskName[Source: Sequence Source -> Map -> Sink: Print to Std. Out]
2> 2
1> 6
map()-----subtask:1--taskName[Source: Sequence Source -> Map -> Sink: Print to Std. Out]
map()-----subtask:0--taskName[Source: Sequence Source -> Map -> Sink: Print to Std. Out]
2> 3
1> 7
map()-----subtask:1--taskName[Source: Sequence Source -> Map -> Sink: Print to Std. Out]
2> 4
map()-----subtask:1--taskName[Source: Sequence Source -> Map -> Sink: Print to Std. Out]
map()-----subtask:0--taskName[Source: Sequence Source -> Map -> Sink: Print to Std. Out]
2> 5
1> 8
map()-----subtask:0--taskName[Source: Sequence Source -> Map -> Sink: Print to Std. Out]
1> 9
map()-----subtask:0--taskName[Source: Sequence Source -> Map -> Sink: Print to Std. Out]
1> 10
close()-----subtask:1--taskName[Source: Sequence Source -> Map -> Sink: Print to Std. Out]
close()-----subtask:0--taskName[Source: Sequence Source -> Map -> Sink: Print to Std. Out]

根据执行结果可知：

2个并行度，所以 2个任务
每个任务先执行1次 open()
有多少个数据，就执行多少次 map() 方法
最后才执行 close() 方法

参考

https://blog.csdn.net/Yuan_CSDF/article/details/117375805
https://blog.csdn.net/qq_33982605/article/details/106286917

原文出处：/show_1IX45rHe2bEq.html