Flink 几种checkpoint的姿势，脚本，web端，zeppelin

秒速五厘米 2022-11-25 05:24 170阅读 0赞

1，主要验证了一下flink on zeppelin checkpoint的这个功能，临时写了个案例去测试一下。

2，首先我们知道正常的checkpoint恢复取消的任务，我们需要在脚本里面指定路径，这个我之前博客有写，直接贴官网的案例：

[https://ci.apache.org/projects/flink/flink-docs-release-1.11/zh/ops/state/checkpoints.html][https_ci.apache.org_projects_flink_flink-docs-release-1.11_zh_ops_state_checkpoints.html]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70][]

3，不知道大家是不是这样子，如果在一些集群环境运行。我们的任务需要重启，需要登录服务器，重新上传jar包，再执行脚本，非常的繁琐。

还好 Flink提供了web端上传jar执行命令，这个真的好用：

我们上传jar ，然后指定主类，需要传入的参数，并行度。代码里面已经指定了checkpoint的路径，第一次执行不需要指定。然后提交任务：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 1][]

代码如下：

>     package com.test
>     
>     import java.util.Calendar
>     import org.apache.flink.api.common.restartstrategy.RestartStrategies
>     import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
>     import org.apache.flink.api.common.time.Time
>     import org.apache.flink.api.scala.typeutils.Types
>     import org.apache.flink.contrib.streaming.state.RocksDBStateBackend
>     import org.apache.flink.runtime.state.StateBackend
>     import org.apache.flink.streaming.api.environment.CheckpointConfig
>     import org.apache.flink.streaming.api.functions.KeyedProcessFunction
>     import org.apache.flink.streaming.api.functions.source.{RichSourceFunction, SourceFunction}
>     import org.apache.flink.streaming.api.scala._
>     import org.apache.flink.streaming.api.{CheckpointingMode, TimeCharacteristic}
>     import org.apache.flink.util.Collector
>     
>     import scala.collection.immutable
>     import scala.util.Random
>     
>     /*
>      *  todo 测试
>      *   senv 是zeppelin的写法，目前要统一。
>      *  com.test.CheckpointCancelTest
>      */
>     object ZeppelinCheckpointCancelTest {
>     
>     
>       def main(args: Array[String]): Unit = {
>         val senv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
>         senv.setParallelism(1)
>         senv.setStreamTimeCharacteristic(TimeCharacteristic.EventTime) //设置时间模式为事件时间
>         senv.enableCheckpointing(60000l) //1分钟做一次checkpoint
>         val checkpointConfig = senv.getCheckpointConfig
>         checkpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE) //仅仅一次
>         checkpointConfig.setMinPauseBetweenCheckpoints(30000l) //设置checkpoint间隔时间30秒
>         checkpointConfig.setCheckpointTimeout(100000l) //设置checkpoint超时时间
>         checkpointConfig.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION) //cancel时保留checkpoint
>     
>         //设置statebackend 为rockdb
>         val stateBackend: StateBackend = new RocksDBStateBackend("hdfs://dev-ct6-dc-master01:8020/flink-1.11.0/checkpoints/rocksDBStateBackend")
>         senv.setStateBackend(stateBackend)
>         //设置重启策略   重启3次 间隔10秒
>         senv.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, Time.seconds(10)))
>     
>         senv.addSource(new RichSourceFunction[(String, Double)] {
>           override def run(sourceContext: SourceFunction.SourceContext[(String, Double)]): Unit = {
>             val random = new Random()
>     
>             val taskIdx: Int = this.getRuntimeContext.getIndexOfThisSubtask
>     
>             var curFtemp: immutable.IndexedSeq[(String, Double)] = (1 to 2).map {
>               i => ("sensor_" + (taskIdx * 10 + i), 65 + (random.nextGaussian() * 20))
>             }
>     
>             while (true) {
>               curFtemp = curFtemp.map(t => (t._1, t._2 + (random.nextGaussian() * 0.5)))
>               val ts: Long = Calendar.getInstance.getTimeInMillis
>               curFtemp.foreach(t => sourceContext.collect((t._1, t._2)))
>               Thread.sleep(1000)
>             }
>           }
>     
>           override def cancel(): Unit = {
>     
>           }
>         }).keyBy(_._1)
>           .process(new KeyedProcessFunction[String, (String,Double), String] {
>             // 保存上一个传感器温度
>             lazy val lastTemp: ValueState[Double] = getRuntimeContext
>               .getState(new ValueStateDescriptor[Double]("lastTemp", Types.of[Double]))
>     
>     
>             override def processElement(value: (String,Double),
>                                         ctx: KeyedProcessFunction[String, (String,Double), String]#Context,
>                                         out: Collector[String]): Unit = {
>     
>               var preTemp: Double = lastTemp.value()
>               if (preTemp % 7 == 0) {
>                 println("搞死美国preTemp= " + preTemp)
>               }
>               if (preTemp == 0.0) {
>                 preTemp = 100
>                 lastTemp.update(preTemp)
>               } else {
>                 preTemp = preTemp - 1
>                 lastTemp.update(preTemp)
>               }
>             }
>           })
>           .print()
>         senv.execute("CheckpointCancelTest")
>       }
>     
>     }

我们在web端看见，checkpoint起到作用了：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 2][]

并且我们看到日志打印的state：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 3][]

在红框框的时候正好是对应chk-17，我们再恢复任务的时候，会从它开始消费。

关闭任务，再次填入checkpoint的地址，我们提交任务：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 4][]

任务正常执行之后我们发现日志打印是从最后一次的保存state打印：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 5][]

4，同样的代码我们在zeppelin怎么做到checkpoint恢复任务呢。

1) 肯定先要有zeppelin环境

2）代码还是那套代码，稍微有点点变动，如果觉得麻烦，可以直接写带状态的SQL，至于什么样子的SQL是会使用state的？

我好像之前在群里问过，一下子想不起来了，依稀记得官网sql部分有提到。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 6][]

> %flink
> 
> import java.util.Calendar  
> import org.apache.flink.api.common.restartstrategy.RestartStrategies  
> import org.apache.flink.api.common.state.\{ValueState, ValueStateDescriptor\}  
> import org.apache.flink.api.common.time.Time  
> import org.apache.flink.api.scala.typeutils.Types  
> import org.apache.flink.contrib.streaming.state.RocksDBStateBackend  
> import org.apache.flink.runtime.state.StateBackend  
> import org.apache.flink.streaming.api.environment.CheckpointConfig  
> import org.apache.flink.streaming.api.functions.KeyedProcessFunction  
> import org.apache.flink.streaming.api.functions.source.\{RichSourceFunction, SourceFunction\}  
> import org.apache.flink.streaming.api.scala.\_  
> import org.apache.flink.streaming.api.\{CheckpointingMode, TimeCharacteristic\}  
> import org.apache.flink.util.Collector
> 
> import scala.collection.immutable  
> import scala.util.Random
> 
>    
>     senv.setParallelism(5)  
>     senv.setStreamTimeCharacteristic(TimeCharacteristic.EventTime) //设置时间模式为事件时间  
>     senv.enableCheckpointing(60000l) //1分钟做一次checkpoint  
>     val checkpointConfig = senv.getCheckpointConfig  
>     checkpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY\_ONCE) //仅仅一次  
>     checkpointConfig.setMinPauseBetweenCheckpoints(30000l) //设置checkpoint间隔时间30秒  
>     checkpointConfig.setCheckpointTimeout(100000l) //设置checkpoint超时时间  
>     checkpointConfig.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN\_ON\_CANCELLATION) //cancel时保留checkpoint
> 
>     //设置statebackend 为rockdb  
>     val stateBackend: StateBackend = new RocksDBStateBackend("hdfs://dev-ct6-dc-master01:8020/flink-1.11.0/checkpoints/rocksDBStateBackend")  
>     senv.setStateBackend(stateBackend)  
>     //设置重启策略   重启3次 间隔10秒  
>     senv.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, Time.seconds(10)))
> 
>     senv.addSource(new RichSourceFunction\[(String, Double)\] \{  
>       override def run(sourceContext: SourceFunction.SourceContext\[(String, Double)\]): Unit = \{  
>         val random = new Random()
> 
>         val taskIdx: Int = this.getRuntimeContext.getIndexOfThisSubtask
> 
>         var curFtemp: immutable.IndexedSeq\[(String, Double)\] = (1 to 2).map \{  
>           i => ("sensor\_" + (taskIdx \* 10 + i), 65 + (random.nextGaussian() \* 20))  
>         \}
> 
>         while (true) \{  
>           curFtemp = curFtemp.map(t => (t.\_1, t.\_2 + (random.nextGaussian() \* 0.5)))  
>           val ts: Long = Calendar.getInstance.getTimeInMillis  
>           curFtemp.foreach(t => sourceContext.collect((t.\_1, t.\_2)))  
>           Thread.sleep(1000)  
>         \}  
>       \}
> 
>       override def cancel(): Unit = \{
> 
>       \}  
>     \}).keyBy(\_.\_1)  
>       .process(new KeyedProcessFunction\[String, (String,Double), String\] \{  
>         // 保存上一个传感器温度  
>         lazy val lastTemp: ValueState\[Double\] = getRuntimeContext  
>           .getState(new ValueStateDescriptor\[Double\]("lastTemp", Types.of\[Double\]))
> 
>   
>         override def processElement(value: (String,Double),  
>                                     ctx: KeyedProcessFunction\[String, (String,Double), String\]\#Context,  
>                                     out: Collector\[String\]): Unit = \{
> 
>           var preTemp: Double = lastTemp.value()  
>           if (preTemp % 7 == 0) \{  
>             println("搞死美国preTemp= " + preTemp)  
>           \}  
>           if (preTemp == 0.0) \{  
>             preTemp = 100  
>             lastTemp.update(preTemp)  
>           \} else \{  
>             preTemp = preTemp - 1  
>             lastTemp.update(preTemp)  
>           \}  
>         \}  
>       \})  
>       .print()  
>     senv.execute("CheckpointCancelTest")  
>

3）任务运行之后停止，然后

加入flink 环境配置。**重启该Notebook的Flink Interpreter，随后重新运行%flink.conf使得新配置生效**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 7][]

然后启动

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 8][]

查看日志，从checkpoint恢复state了：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 9][]

总结：

毕竟scala代码在zeppelin执行不够优雅，一般都是写SQL的，只能说我是验证一下，本身zeppelin对scala是支持的很好的。

可能会存在一些问题的，不过在群里面 大佬会帮忙解答：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 10][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 11][]

那搞zeppelin必须要看 文档：[https://www.yuque.com/jeffzhangjianfeng/gldg8w/bam5y1][https_www.yuque.com_jeffzhangjianfeng_gldg8w_bam5y1]

简锋大佬棒棒的：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 12][]

最后还是好文章要分享一下：

先跟鸡哥打个广告 ，博客地址: [https://me.csdn.net/weixin\_47482194][https_me.csdn.net_weixin_47482194]

写的博客很有水平的，上了几次官网推荐了。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 13][]

[https_ci.apache.org_projects_flink_flink-docs-release-1.11_zh_ops_state_checkpoints.html]: https://ci.apache.org/projects/flink/flink-docs-release-1.11/zh/ops/state/checkpoints.html
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70]: /images/20221124/e3abd8e8164f4bbd92c4d3e07839376f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 1]: /images/20221124/50762dfa490b495391c9caa001f5e5e9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 2]: /images/20221124/058b6a65acad4af7b6481a7dbb6a8b47.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 3]: /images/20221124/421919c2284643cfa24f64196c94dc56.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 4]: /images/20221124/48ce898ce0b140819e9b915f9c234c7d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 5]: /images/20221124/bd777f07a70b444cb112104a5eaa8a46.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 6]: /images/20221124/62a9a7539a134d71af7355f5f9e21bc0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 7]: /images/20221124/bc68071be4d74a8199899e18bfe6acdd.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 8]: /images/20221124/a8a8acc67dbe493cb0f2d39b37d79eb7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 9]: /images/20221124/e84f2ca5a981405db27daedc121e3cc8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 10]: /images/20221124/99c25a6e1274436dbe6bc008aee8df9d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 11]: /images/20221124/eb79d9505a08483d944396dc87eed971.png
[https_www.yuque.com_jeffzhangjianfeng_gldg8w_bam5y1]: https://www.yuque.com/jeffzhangjianfeng/gldg8w/bam5y1
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 12]: /images/20221124/3ef6fc1691d64da7ad98ec29c19f5a64.png
[https_me.csdn.net_weixin_47482194]: https://me.csdn.net/weixin_47482194
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 13]: /images/20221124/c52cf945b9f14416bf684a35f4e30248.png