Flink State |检查最后一个数据是第一个数据

import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

public class FirstLastCheck extends KeyedProcessFunction<String, String, String> {

    private transient ValueState<String> firstDataState;

    @Override
    public void open(Configuration parameters) {
        ValueStateDescriptor<String> descriptor =
                new ValueStateDescriptor<>("firstData", String.class);
        firstDataState = getRuntimeContext().getState(descriptor);
    }

    @Override
    public void processElement(String value, Context ctx, Collector<String> out) throws Exception {
        if (firstDataState.value() == null) {
            firstDataState.update(value);
            out.collect("First data: " + value);
        } else {
            String firstData = firstDataState.value();
            if (value.equals(firstData)) {
                out.collect("Last data is the same as first data: " + value);
            } else {
                out.collect("Last data is different from first data: " + value);
            }
        }
    }
}