2.3?算子狀態(tài)

相比于鍵控狀態(tài)，算子狀態(tài)目前支持的存儲類型只有以下三種：

• ListState：存儲列表類型的狀態(tài)。
• UnionListState：存儲列表類型的狀態(tài)，與?ListState?的區(qū)別在于：如果并行度發(fā)生變化，ListState?會將該算子的所有并發(fā)的狀態(tài)實例進行匯總，然后均分給新的?Task；而?UnionListState?只是將所有并發(fā)的狀態(tài)實例匯總起來，具體的劃分行為則由用戶進行定義。
• BroadcastState：用于廣播的算子狀態(tài)。

這里我們繼續(xù)沿用上面的例子，假設(shè)此時我們不需要區(qū)分監(jiān)控數(shù)據(jù)的類型，只要有監(jiān)控數(shù)據(jù)超過閾值并達(dá)到指定的次數(shù)后，就進行報警，代碼如下：

public class ThresholdWarning extends RichFlatMapFunction<Tuple2<String, Long>, 
Tuple2<String, List<Tuple2<String, Long>>>> implements CheckpointedFunction {

    // 非正常數(shù)據(jù)
    private List<Tuple2<String, Long>> bufferedData;
    // checkPointedState
    private transient ListState<Tuple2<String, Long>> checkPointedState;
    // 需要監(jiān)控的閾值
    private Long threshold;
    // 次數(shù)
    private Integer numberOfTimes;

    ThresholdWarning(Long threshold, Integer numberOfTimes) {
        this.threshold = threshold;
        this.numberOfTimes = numberOfTimes;
        this.bufferedData = new ArrayList<>();
    }

    @Override
    public void initializeState(FunctionInitializationContext context) throws Exception {
        // 注意這里獲取的是OperatorStateStore
        checkPointedState = context.getOperatorStateStore().
            getListState(new ListStateDescriptor<>("abnormalData",
                TypeInformation.of(new TypeHint<Tuple2<String, Long>>() {
                })));
        // 如果發(fā)生重啟，則需要從快照中將狀態(tài)進行恢復(fù)
        if (context.isRestored()) {
            for (Tuple2<String, Long> element : checkPointedState.get()) {
                bufferedData.add(element);
            }
        }
    }

    @Override
    public void flatMap(Tuple2<String, Long> value, 
                        Collector<Tuple2<String, List<Tuple2<String, Long>>>> out) {
        Long inputValue = value.f1;
        // 超過閾值則進行記錄
        if (inputValue >= threshold) {
            bufferedData.add(value);
        }
        // 超過指定次數(shù)則輸出報警信息
        if (bufferedData.size() >= numberOfTimes) {
             // 順便輸出狀態(tài)實例的hashcode
             out.collect(Tuple2.of(checkPointedState.hashCode() + "閾值警報！", bufferedData));
            bufferedData.clear();
        }
    }

    @Override
    public void snapshotState(FunctionSnapshotContext context) throws Exception {
        // 在進行快照時，將數(shù)據(jù)存儲到checkPointedState
        checkPointedState.clear();
        for (Tuple2<String, Long> element : bufferedData) {
            checkPointedState.add(element);
        }
    }
}

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 開啟檢查點機制
env.enableCheckpointing(1000);
// 設(shè)置并行度為1
DataStreamSource<Tuple2<String, Long>> tuple2DataStreamSource = env.setParallelism(1).fromElements(
    Tuple2.of("a", 50L), Tuple2.of("a", 80L), Tuple2.of("a", 400L),
    Tuple2.of("a", 100L), Tuple2.of("a", 200L), Tuple2.of("a", 200L),
    Tuple2.of("b", 100L), Tuple2.of("b", 200L), Tuple2.of("b", 200L),
    Tuple2.of("b", 500L), Tuple2.of("b", 600L), Tuple2.of("b", 700L));
tuple2DataStreamSource
    .flatMap(new ThresholdWarning(100L, 3))
    .printToErr();
env.execute("Managed Keyed State");
}

//?開啟檢查點機制，并指定狀態(tài)檢查點之間的時間間隔
env.enableCheckpointing(1000); 

//?其他可選配置如下：
//?設(shè)置語義
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
//?設(shè)置兩個檢查點之間的最小時間間隔
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);
//?設(shè)置執(zhí)行Checkpoint操作時的超時時間
env.getCheckpointConfig().setCheckpointTimeout(60000);
//?設(shè)置最大并發(fā)執(zhí)行的檢查點的數(shù)量
env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
//?將檢查點持久化到外部存儲
env.getCheckpointConfig().enableExternalizedCheckpoints(
????ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
//?如果有更近的保存點時，是否將作業(yè)回退到該檢查點
env.getCheckpointConfig().setPreferCheckpointForRecovery(true);

#?觸發(fā)指定id的作業(yè)的Savepoint，并將結(jié)果存儲到指定目錄下
bin/flink?savepoint?:jobId?[:targetDirectory]

//?配置?FsStateBackend
env.setStateBackend(new?FsStateBackend("hdfs://namenode:40010/flink/checkpoints"));
//?配置?RocksDBStateBackend
env.setStateBackend(new?RocksDBStateBackend("hdfs://namenode:40010/flink/checkpoints"));

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-statebackend-rocksdb_2.11</artifactId>
    <version>1.14.2</version>
</dependency>

state.backend:?filesystem
state.checkpoints.dir:?hdfs://namenode:40010/flink/checkpoints