flink checkpoint »Ö¸´_Flink ´ÓCheckpointÖлָ´×÷Òµ

Flink 1.11 °æ±¾

1. ÅäÖÃ

Èç¹ûÎÒÃǵÄÈÎÎñÒѾ­Ö´Ðкܳ¤Ê±¼ä，ͻȻÓöµ½¹ÊÕÏÍ£Ö¹，ÄÇôÖмä¹ý³Ì´¦Àí½á¹û¾Í»áÈ«²¿¶ªÊ§，ÖØÆôºóÐèÒªÖØдÓÉÏÒ»´Î¿ªÊ¼µÄλÖÃÏû·Ñ，Õâ»á»¨·ÑÎÒÃǺܳ¤µÄʱ¼ä¡£ÕâÖÖ½á¾ÖÏÔʾÎÒÃDz»ÄܽÓÊÜ，ÎÒÃÇÏ£ÍûµÄÊÇ×÷ÒµÔÚ¹ÊÕÏʧ°ÜÖØÆôºóÄܱ£Áô֮ǰµÄ״̬²¢ÄÜ´Óʧ°ÜµÄλÖüÌÐøÏû·Ñ¡£¿ÉÒÔͨ¹ýÈçÏÂÅäÖñ£´æ´¦Àí״̬：

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();// ¿ªÆôCheckpointenv.enableCheckpointing(1000);// ÉèÖÃ״̬ºó¶Ëenv.setStateBackend(new FsStateBackend("hdfs://localhost:9000/flink/checkpoint"));env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);env.getCheckpointConfig().setCheckpointTimeout(60000);

×÷ҵֹͣºó CheckPoint Êý¾ÝĬÈÏ»á×Ô¶¯É¾³ý，ËùÒÔÐèÒªÈçÏÂÅäÖÃÀ´ÉèÖÃÔÚ×÷ҵʧ°Ü±»È¡Ïûºó CheckPoint Êý¾Ý²»±»É¾³ý：

env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

2. ÑéÖ¤

ÎÒÃÇʹÓþ­µäµÄ WordCount ʵÀýÀ´ÑéÖ¤´Ó Checkpoint Öлָ´×÷Òµ²¢ÄÜÑØÓÃ֮ǰµÄ״̬ÐÅÏ¢¡£ÎªÁËÄ£Äâ×÷ҵʧ°Ü²¢Äָܻ´，ÎÒÃÇÅжϵ±ÎÒÃÇÊäÈëÊÇ ¡°ERROR¡± ʱ，Å׳öÒì³£ÆÈʹ×÷ҵʧ°Ü：

public void flatMap(String value, Collector out) {    // ʧ°ÜÐźŠ   if (Objects.equals(value, "ERROR")) {        throw new RuntimeException("custom error flag, restart application");    }    ...}

ΪÁËÈ·±£×÷ÒµÔÚʧ°ÜºóÄÜ×Ô¶¯»Ö¸´，ÎÒÃÇÉèÖÃÁËÖØÆô²ßÂÔ，ʧ°Üºó×î¶àÖØÆô3´Î，ÿ´ÎÖØÆô¼ä¸ô10s：

env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 10000));

ÎÒÃÇ¿´Ò»ÏÂÏêϸµÄ´úÂë：

public class RestoreCheckpointExample {    public static void main(String[] args) throws Exception {        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();        // ÅäÖÃCheckpoint        env.enableCheckpointing(1000);        env.setStateBackend(new FsStateBackend("hdfs://localhost:9000/flink/checkpoint"));        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);        env.getCheckpointConfig().setCheckpointTimeout(60000);        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);        // ÅäÖÃʧ°ÜÖØÆô²ßÂÔ：ʧ°Üºó×î¶àÖØÆô3´Î ÿ´ÎÖØÆô¼ä¸ô10s        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 10000));        DataStream source = env.socketTextStream("localhost", 9100, "\n")                .name("MySourceFunction");        DataStream> wordsCount = source.flatMap(new FlatMapFunction>() {            @Override            public void flatMap(String value, Collector out) {                // ʧ°ÜÐźŠ               if (Objects.equals(value, "ERROR")) {                    throw new RuntimeException("custom error flag, restart application");                }                // ²ð·Öµ¥´Ê                for (String word : value.split("\\s")) {                    out.collect(Tuple2.of(word, 1));                }            }        }).name("MyFlatMapFunction");        DataStream> windowCount = wordsCount                .keyBy(new KeySelector, String>() {                    @Override                    public String getKey(Tuple2 tuple) throws Exception {                        return tuple.f0;                    }                })                .sum(1).name("MySumFunction");        windowCount.print().setParallelism(1).name("MyPrintFunction");        env.execute("RestoreCheckpointExample");    }}

´úÂëµØÖ·

ÏÂÃæÎÒÃǾßÌå²Ù×÷½øÐÐÑéÖ¤¡£Ê×ÏÈÆô¶¯Ò»¸ö nc ·þÎñ：

wy:opt wy$ nc -lk 9100

¶Ë¿ÚºÅΪ：9100

È»ºóÆô¶¯ RestoreCheckpointExample ×÷Òµ：

wy:~ wy$ flink run -c com.flink.example.stream.state.checkpoint.RestoreCheckpointExample  ~/study/code/data-example/flink-example/target/flink-example-1.0.jar

ϱíÊÇ´Ó nc ·þÎñÊä³ö²âÊÔÊý¾Ý，´Ó Flink Web Ò³ÃæÊä³ö½á¹ûÊý¾ÝµÄÏêϸÐÅÏ¢：

´ÓÉÏÃæÐÅÏ¢¿ÉÒÔ¿´³ö×÷Òµ»Ö¸´ºó，¼ÆËã½á¹ûÒ²ÊÇ»ùÓÚ×÷ҵʧ°ÜÇ°±£´æµÄ״̬ÉϼÆËãµÄ¡£ÎÒÃÇÉèÖÃ×î¶à¿ÉÒÔÖØÆôÈý´Î，µ±ÎÒÃǵÚËÄ´ÎÊäÈë ¡°ERROR¡± Êý¾Ýʱ，³ÌÐò³¹µ×ʧ°Ü¡£

3. ×÷ҵ״̬±ä»¯

·¢ËÍ ERROR Ðźźó，flatMap Ëã×ÓÅ׳öÒì³£，ÓÉ RUNNING ״̬Çл»Îª FAILED，µ¼ÖÂ×÷Òµ±»È¡Ïû：

2020-12-26 20:48:12,967 WARN  org.apache.flink.runtime.taskmanager.Task                    [] - Source: MySourceFunction -> MyFlatMapFunction (1/1) (be8abffb0f6815889929dc9b605b7ae5) switched from RUNNING to FAILED.java.lang.RuntimeException: custom error flag, restart application at com.flink.example.stream.state.checkpoint.RestoreCheckpointExample$1.flatMap(RestoreCheckpointExample.java:39) ~[blob_p-353721c11ae1acd403dc8be3b663e9a60854d5c3-b6237955a73f418e6d7b272281b64594:?] at com.flink.example.stream.state.checkpoint.RestoreCheckpointExample$1.flatMap(RestoreCheckpointExample.java:34) ~[blob_p-353721c11ae1acd403dc8be3b663e9a60854d5c3-b6237955a73f418e6d7b272281b64594:?] at org.apache.flink.streaming.api.operators.StreamFlatMap.processElement(StreamFlatMap.java:50) ~[flink-dist_2.12-1.11.2.jar:1.11.2] at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.pushToOperator(OperatorChain.java:717) ~[flink-dist_2.12-1.11.2.jar:1.11.2] at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:692) ~[flink-dist_2.12-1.11.2.jar:1.11.2] at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:672) ~[flink-dist_2.12-1.11.2.jar:1.11.2] at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:52) ~[flink-dist_2.12-1.11.2.jar:1.11.2] at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:30) ~[flink-dist_2.12-1.11.2.jar:1.11.2] at org.apache.flink.streaming.api.operators.StreamSourceContexts$NonTimestampContext.collect(StreamSourceContexts.java:104) ~[flink-dist_2.12-1.11.2.jar:1.11.2] at org.apache.flink.streaming.api.functions.source.SocketTextStreamFunction.run(SocketTextStreamFunction.java:111) ~[flink-dist_2.12-1.11.2.jar:1.11.2] at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:100) ~[flink-dist_2.12-1.11.2.jar:1.11.2] at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:63) ~[flink-dist_2.12-1.11.2.jar:1.11.2] at org.apache.flink.streaming.runtime.tasks.SourceStreamTask$LegacySourceFunctionThread.run(SourceStreamTask.java:213) ~[flink-dist_2.12-1.11.2.jar:1.11.2]2020-12-26 20:48:12,978 INFO  org.apache.flink.runtime.taskmanager.Task                    [] - Attempting to cancel task MySumFunction -> Sink: MyPrintFunction (1/1) (d464321ae464046684fd28d37bdcc3d7).2020-12-26 20:48:12,978 INFO  org.apache.flink.runtime.taskmanager.Task                    [] - MySumFunction -> Sink: MyPrintFunction (1/1) (d464321ae464046684fd28d37bdcc3d7) switched from RUNNING to CANCELING....2020-12-26 20:48:12,979 INFO  org.apache.flink.runtime.taskmanager.Task                    [] - MySumFunction -> Sink: MyPrintFunction (1/1) (d464321ae464046684fd28d37bdcc3d7) switched from CANCELING to CANCELED.

ÓÉÓÚÎÒÃÇÉèÖÃÁËÖØÆô²ßÂÔ，ÖØÆô¼ä¸ôΪ10s，ËùÒÔ×÷ÒµÔÚ10sÖ®ºóÖØÆô，¾­¹ý CREATED -> DEPLOYING -> RUNNING ״̬，×÷Òµ±»ÖØÆô：

2020-12-26 20:48:22,997 INFO  org.apache.flink.runtime.taskmanager.Task                    [] - Source: MySourceFunction -> MyFlatMapFunction (1/1) (223b777dfc69013852e9ab37d3cc078e) switched from CREATED to DEPLOYING....2020-12-26 20:48:22,998 INFO  org.apache.flink.runtime.taskmanager.Task                    [] - Source: MySourceFunction -> MyFlatMapFunction (1/1) (223b777dfc69013852e9ab37d3cc078e) switched from DEPLOYING to RUNNING.2020-12-26 20:48:22,999 INFO  org.apache.flink.runtime.taskmanager.Task                    [] - MySumFunction -> Sink: MyPrintFunction (1/1) (53e45aa6b16f0b82d1bde8325f0cfbaf) switched from CREATED to DEPLOYING....2020-12-26 20:48:23,000 INFO  org.apache.flink.runtime.taskmanager.Task                    [] - MySumFunction -> Sink: MyPrintFunction (1/1) (53e45aa6b16f0b82d1bde8325f0cfbaf) switched from DEPLOYING to RUNNING.

ÓÉÓÚÎÒÃÇÉèÖÃÁË×î¶àÖØÆôÈý´Î，ËùÒÔµÚËĴη¢³ö ERROR Ðźźó，×÷Òµ³¹µ×ʧ°Ü：

2020-12-26 21:05:29,294 WARN  org.apache.flink.runtime.taskmanager.Task                    [] - Source: MySourceFunction -> MyFlatMapFunction (1/1) (223b777dfc69013852e9ab37d3cc078e) switched from RUNNING to FAILED.java.lang.RuntimeException: custom error flag, restart application at com.flink.example.stream.state.checkpoint.RestoreCheckpointExample$1.flatMap(RestoreCheckpointExample.java:39) ~[blob_p-353721c11ae1acd403dc8be3b663e9a60854d5c3-b6237955a73f418e6d7b272281b64594:?] at com.flink.example.stream.state.checkpoint.RestoreCheckpointExample$1.flatMap(RestoreCheckpointExample.java:34) ~[blob_p-353721c11ae1acd403dc8be3b663e9a60854d5c3-b6237955a73f418e6d7b272281b64594:?] at org.apache.flink.streaming.api.operators.StreamFlatMap.processElement(StreamFlatMap.java:50) ~[flink-dist_2.12-1.11.2.jar:1.11.2]...2020-12-26 21:05:29,332 INFO  org.apache.flink.runtime.taskmanager.Task                    [] - Attempting to cancel task MySumFunction -> Sink: MyPrintFunction (1/1) (53e45aa6b16f0b82d1bde8325f0cfbaf).2020-12-26 21:05:29,332 INFO  org.apache.flink.runtime.taskmanager.Task                    [] - MySumFunction -> Sink: MyPrintFunction (1/1) (53e45aa6b16f0b82d1bde8325f0cfbaf) switched from RUNNING to CANCELING....2020-12-26 21:05:29,334 INFO  org.apache.flink.runtime.taskmanager.Task                    [] - MySumFunction -> Sink: MyPrintFunction (1/1) (53e45aa6b16f0b82d1bde8325f0cfbaf) switched from CANCELING to CANCELED....2020-12-26 21:05:29,353 INFO  org.apache.flink.runtime.taskexecutor.DefaultJobLeaderService [] - Remove job a78621726e80e5bde6f936a177f0d052 from job leader monitoring.2020-12-26 21:05:29,353 INFO  org.apache.flink.runtime.taskexecutor.TaskExecutor           [] - Close JobManager connection for job a78621726e80e5bde6f936a177f0d052.

»¶Ó­¹Ø×¢ÎҵĹ«ÖںźͲ©¿Í：