Flink function (1): rich function

rich function, namely “ rich ” function , That is, compared with the general function Come on , It provides more functions . It defines some methods of function life cycle , And you can get the context of the function runtime . There are methods open, close, getRuntimeContext, setRuntimeContext. What is particularly useful is that we can get keyed state, To do more control , such as keyed State Of TTL Set up .（State It's using RuntimeContext Access to the , Therefore, it can only be used in rich functions ）

All transformations that require user-defined functions can convert rich Function as parameter （All transformations that require a user-defined function can instead take as argument a rich function. ） because RichFunction extends Function.

The following is a description of some useful places in the official documents ：

Rich functions provide, in addition to the user-defined function (map, reduce, etc), four methods: open, close, getRuntimeContext, and setRuntimeContext. These are useful for parameterizing the function (see Passing Parameters to Functions), creating and finalizing local state, accessing broadcast variables (see Broadcast Variables), and for accessing runtime information such as accumulators and counters (see Accumulators and Counters), and information on iterations (see Iterations).

The key functions are described below ：

void open(Configuration parameters) throws Exception;

In function （function, such as map join etc. ） Before calling ,open() Method （method） First called , Used to initialize the operation , Therefore, it is suitable for one-time setting . As part of the function iteration , This method will be called before each upper iteration （For functions that are part of an iteration, this method will be invoked at the beginning of each iteration uperstep.）.

void close() throws Exception;

Is the last method called in the life cycle , Do some cleaning .

RuntimeContext getRuntimeContext();

Get runtime context

void setRuntimeContext(RuntimeContext t);

Set runtime context . Each parallel operator subtask has a runtime context , The context records some information about the operation of this operator , Including the current parallelism of the operator 、 Operator subtask sequence number 、 Broadcast data 、 accumulator 、 Monitoring data . most important of all , We can get state data from context ;

Example implementation Count Of TTL Set up

package com.xx.bigdata.flink;

import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;

public class TTLCountMapFunction extends RichMapFunction<Tuple2<String, Integer>, Tuple2<String, Integer>>  {

    private transient ValueState<Integer> state;

    @Override
    public Tuple2<String, Integer> map(Tuple2<String, Integer> in) throws Exception {
        Integer currentCount = state.value();
        if(null == currentCount) {
            currentCount = 0;
        }
        Integer nowCount = currentCount + in.f1;
        state.update(nowCount);
        return new Tuple2<>(in.f0, nowCount);
    }

    public void open(Configuration parameters) throws Exception {
        StateTtlConfig ttlConfig = StateTtlConfig
                .newBuilder(Time.seconds(300))
                .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
                .build();

        ValueStateDescriptor<Integer> stateDescriptor = new ValueStateDescriptor<>("count-state", Integer.class);
        stateDescriptor.enableTimeToLive(ttlConfig);
        state = getRuntimeContext().getState(stateDescriptor);
    }
}