Flink之state processor api實踐

    

    前不久，Flink社區發布了FLink 1.9版本，在其中包含了一個很重要的新特性，即state processor api，這個框架支持對checkpoint和savepoint進行操作，包括讀取、變更、寫入等等。

savepoint的可操作帶來了很多的可能性：

• 作業遷移

　　1.跨類型作業，假如有一個storm作業，將狀態緩存在外部系統，希望更好的利用flink的狀態機制來增加作業的穩定和減少數據的延遲，但如果直接遷移，必然面臨狀態的丟失，這時，可以將外部系統的狀態轉換為flink作業的savepoint來啟動。
　　2.同類型作業，假如有一個flink作業已經在運行，一個新的flink作業希望復用之前的某些狀態，也可以將savepoint進行處理重新寫入，進而使得新的flink作業可以在某個基礎上運行。

• 作業升級

　　1.有UID升級，一般情況下，如果升級前的operator已經設置了uid，那么可以直接升級，但是如果希望在之前的狀態數據上做些變更，這里就提供了一種接口。
　　2.無UID升級，在特殊情況下，一開始編寫了沒有UID的作業，后來改成了標准的有UID的作業，反而無法在之前的savepoint上啟動了，這時也可以對savepoint同時做升級。

• 作業校驗

　　1.異步校驗，一般而言，flink作業的最終結果都會持久化輸出，但在面臨問題的時候，如何確定哪一級出現問題，state processor api也提供了一種可能，去檢驗state中的數據是否與預期的一致。

• 作業擴展

　　1.橫向擴展，如果在flink作業一開始運行的時候，因為面對的數據量較小，設置了比較小的最大並行度，但在數據量增大的時候，卻沒辦法從老的savepoint以一個比之前的最大並行度更大的並行度來啟動作業，這時，也需要復寫savepoint的同時更改最大並行度。
　　2.縱向擴展，在flink作業中新添加了一個operator，從savepoint啟動的時候這個operator默認無狀態，可以手動構造數據，使得這個operator的表現和其他operator保持一致。

可以對savepoint進行哪些操作？

• 讀取savepoint

　　1.驗證，讀取出來的savepoint會轉換為一個dataSet，隨后可以以標准批處理的方式來驗證你的業務預期；
　　2.source，也可以以savepoint作為數據源，來作為你另一個作業的輸入。

• 寫入savepoint

　　1.寫入新的savepoint，可以寫入一個全新的savepoint，這個savepoint是獨立的存在，他可以有新的operator uid，新的operator state，以及新的max parallism等等。
　　2.復用原來的savepoint，可以在原來的savepoint的基礎上加入新的operator的state，在新的savepoint被使用之前，老的savepoint不允許被刪除。

那么究竟哪些state是可讀的？有哪些接口了？

可以看到，主要提供對三種state的訪問，operator state和broadcast state，其中broadcast state是一種特殊的operator state，因為他也支持自定義的serializer。

通關程序

目前在社區或者網上並沒有完整的樣例供大家參考，下面這個例子是完全在測試環境中跑通的，所有的flink相關組件的版本依賴都是1.9.0。

下面我們說明如何使用這個框架。

1.首先我們創建一個樣例作業來生成savepoint

主類代碼

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(60*1000);
        DataStream<Tuple2<Integer,Integer>> kafkaDataStream =
                env.addSource(new SourceFunction<Tuple2<Integer,Integer>>() {
            private boolean running = true;
            private int key;
            private int value;
            private Random random = new Random();
            @Override
            public void run(SourceContext<Tuple2<Integer,Integer>> sourceContext) throws Exception {
                while (running){
                    key = random.nextInt(5);
                    sourceContext.collect(new Tuple2<>(key,value++) );
                    Thread.sleep(100);
                }
            }

            @Override
            public void cancel() {
                running = false;
            }
        }).name("source").uid("source");


        kafkaDataStream
                .keyBy(tuple -> tuple.f0)
                .map(new StateTest.StateMap()).name("map").uid("map")
                .print().name("print").uid("print");

在上面的代碼中，只需要注意在自定義的source中，發送tuple2消息，而做savepoint的
關鍵在於狀態，狀態在StateMap這個類中，如下：

public static class StateMap extends RichMapFunction<Tuple2<Integer,Integer>,String>  {
        private transient ListState<Integer> listState;

        @Override
        public void open(Configuration parameters) throws Exception {
            ListStateDescriptor<Integer> lsd =
                    new ListStateDescriptor<>("list",TypeInformation.of(Integer.class));
            listState = getRuntimeContext().getListState(lsd);
        }

        @Override
        public String map(Tuple2<Integer,Integer> value) throws Exception {
            listState.add(value.f1);
            return value.f0+"-"+value.f1;
        }

        @Override
        public void close() throws Exception {
            listState.clear();
        }
    }

在上面的Map中，首先在open中聲明了一個ListState，然后在消息處理的邏輯中，也很簡單的只是把tuple2的值放進了listState中。然后提交作業，等作業運行一段時間之后，觸發一個savepoint，
並記錄savepoint的地址。至此，完成了state processor api驗證工作的數據准備。

2.利用state processor api讀取savepoint
這一步只是簡單驗證下savepoint是否能夠被正確讀取，代碼如下：

public class ReadListState {
    protected static final Logger logger = LoggerFactory.getLogger(ReadListState.class);

    public static void main(String[] args) throws Exception {
        final String operatorUid = "map";
        final String savepointPath =
                "hdfs://xxx/savepoint-41b05d-d517cafb61ba";

        final String checkpointPath = "hdfs://xxx/checkpoints";

        // set up the batch execution environment
        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        RocksDBStateBackend db = new RocksDBStateBackend(checkpointPath);
        DataSet<String> dataSet = Savepoint
                .load(env, savepointPath, db)
                .readKeyedState(operatorUid, new ReaderFunction())
                .flatMap(new FlatMapFunction<KeyedListState, String>() {
                    @Override
                    public void flatMap(KeyedListState keyedListState, Collector<String> collector) throws Exception {
                        keyedListState.value.forEach(new Consumer<Integer>() {
                            @Override
                            public void accept(Integer integer) {
                                collector.collect(keyedListState.key + "-" + integer);
                            }
                        });
                    }
                });

        dataSet.writeAsText("hdfs://xxx/test/savepoint/bravo");

        // execute program
        env.execute("read the list state");
    }

    static class KeyedListState {
        Integer key;
        List<Integer> value;
    }

    static class ReaderFunction extends KeyedStateReaderFunction<Integer, KeyedListState> {
        private transient ListState<Integer> listState;

        @Override
        public void open(Configuration parameters) {
            ListStateDescriptor<Integer> lsd =
                    new ListStateDescriptor<>("list", TypeInformation.of(Integer.class));
            listState = getRuntimeContext().getListState(lsd);
        }

        @Override
        public void readKey(
                Integer key,
                Context ctx,
                Collector<KeyedListState> out) throws Exception {
            List<Integer> li = new ArrayList<>();
            listState.get().forEach(new Consumer<Integer>() {
                @Override
                public void accept(Integer integer) {
                    li.add(integer);
                }
            });

            KeyedListState kl = new KeyedListState();
            kl.key = key;
            kl.value = li;

            out.collect(kl);
        }
    }
}

在讀取了savepoint中的狀態之后，成功將其轉存為一個文件，文件的部分內容如下,每行的內容分別為key-value對：

3.利用state processor api重寫savepoint

savepoint是對程序某個運行時點的狀態的固化，方便程序在再次提交的時候進行接續，但有時候需要對savepoint中的狀態進行改寫，以方便從特定的狀態來啟動作業。

public class ReorganizeListState {
    protected static final Logger logger = LoggerFactory.getLogger(ReorganizeListState.class);
    public static void main(String[] args) throws Exception {
        final String operatorUid = "map";
        final String savepointPath =
                "hdfs://xxx/savepoint-41b05d-d517cafb61ba";

        final String checkpointPath = "hdfs://xxx/checkpoints";

        // set up the batch execution environment
        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        RocksDBStateBackend db = new RocksDBStateBackend(checkpointPath);
        DataSet<KeyedListState> dataSet = Savepoint
                .load(env,savepointPath,db)
                .readKeyedState(operatorUid,new ReaderFunction())
                .flatMap(new FlatMapFunction<KeyedListState, KeyedListState>() {
                    @Override
                    public void flatMap(KeyedListState keyedListState, Collector<KeyedListState> collector) throws Exception {
                        KeyedListState newState = new KeyedListState();
                        newState.value = keyedListState.value.stream()
                        .map( x -> x+10000).collect(Collectors.toList());
                        newState.key = keyedListState.key;
                        collector.collect(newState);
                    }
                });
        
        BootstrapTransformation<KeyedListState> transformation = OperatorTransformation
                .bootstrapWith(dataSet)
                .keyBy(acc -> acc.key)
                .transform(new KeyedListStateBootstrapper());

        Savepoint.create(db,128)
                .withOperator(operatorUid,transformation)
                .write("hdfs://xxx/test/savepoint/");

        // execute program
        env.execute("read the list state");
    }

    static class KeyedListState{
        Integer key;
        List<Integer> value;
    }

    static class ReaderFunction extends KeyedStateReaderFunction<Integer, KeyedListState> {
        private transient ListState<Integer> listState;

        @Override
        public void open(Configuration parameters) {
            ListStateDescriptor<Integer> lsd =
                    new ListStateDescriptor<>("list",TypeInformation.of(Integer.class));
            listState = getRuntimeContext().getListState(lsd);
        }

        @Override
        public void readKey(
                Integer key,
                Context ctx,
                Collector<KeyedListState> out) throws Exception {
            List<Integer> li = new ArrayList<>();
            listState.get().forEach(new Consumer<Integer>() {
                @Override
                public void accept(Integer integer) {
                    li.add(integer);
                }
            });

            KeyedListState kl = new KeyedListState();
            kl.key = key;
            kl.value = li;

            out.collect(kl);
        }
    }

    static class KeyedListStateBootstrapper extends KeyedStateBootstrapFunction<Integer, KeyedListState> {
        private transient ListState<Integer> listState;

        @Override
        public void open(Configuration parameters) {
            ListStateDescriptor<Integer> lsd =
                    new ListStateDescriptor<>("list",TypeInformation.of(Integer.class));
            listState = getRuntimeContext().getListState(lsd);
        }

        @Override
        public void processElement(KeyedListState value, Context ctx) throws Exception {
            listState.addAll(value.value);
        }
    }
}

這里的關鍵在於根據上一步讀取出來dataSet,轉換的過程中將其值全部累加10000，然后將這個dataSet作為輸入來構建一個BootstrapTransformation，然后創建了一個空的savepoint，並把指定
operatorUid的狀態寫為一個savepoint，最終寫入成功，得到了一個新的savepoint，這個新的savepoint包含的狀態中的value相比原先的值發生了變化。

4.驗證新生產的savepoint是否可用

由於驗證用的state是ListState，換言之，是KeyedState，而KeyedState是屬於Flink托管的state，意味着Flink自己掌握狀態的保存和恢復的邏輯，所以為了驗證作業是否正確從新的savepoint
中啟動了，對之前的StateMap改寫如下：

public static class StateMap extends RichMapFunction<Tuple2<Integer,Integer>,String>  {
        private transient ListState<Integer> listState;

        @Override
        public void open(Configuration parameters) throws Exception {
            ListStateDescriptor<Integer> lsd =
                    new ListStateDescriptor<>("list",TypeInformation.of(Integer.class));
            listState = getRuntimeContext().getListState(lsd);
        }

        @Override
        public String map(Tuple2<Integer,Integer> value) throws Exception {
            listState.add(value.f1);
            log.info("get value:{}-{}",value.f0,value.f1);
            StringBuilder sb = new StringBuilder();
            listState.get().forEach(new Consumer<Integer>() {
                @Override
                public void accept(Integer integer) {
                    sb.append(integer).append(";");
                }
            });
            log.info("***********************taskNameAndSubTask:{},restored value:{}"
                    ,getRuntimeContext().getTaskNameWithSubtasks(),sb.toString());
            return value.f0+"-"+value.f1;
        }

        @Override
        public void close() throws Exception {
            listState.clear();
        }
    }

由於無法在state恢復之后立刻就拿到相應恢復的數據，這里之后在每次消息達到的時候輸出下state中的內容，變通的看看是否恢復成功，結果如下：

可以對比看下上圖中key為4的輸出，可以看到輸出的值即為修改后的值，驗證成功。

5.結語
上面我們以一個keyedState來對state processor api做了驗證，但Flink的state分為KeyedState，OperatorState和BroadcastState，在state processor api中都提供相應的處理接口。
另外，對於keyedState，如果作業的並行度發生了變化會如何？如果Key發生了變化會如何？都需要進一步探究。

官方文檔參見：
https://flink.apache.org/feature/2019/09/13/state-processor-api.html
https://ci.apache.org/projects/flink/flink-docs-release-1.9/dev/libs/state_processor_api.html