概述

Flink的exactly-once语义实现是需要依赖checkpoint的,对于一个有状态的Flink任务来说如果想要在任务发生failover,或者手动重启任务的时候任务的状态不丢失是必须要开启checkpoint的,今天这篇文章主要分享一下Flink on zeppelin里面怎么设置checkpoint以及怎么从指定的checkpoint恢复任务.

checkpoint配置

%flink.conf
// 设置任务使用的时间属性是eventtime
pipeline.time-characteristic EventTime
// 设置checkpoint的时间间隔
execution.checkpointing.interval 10000
// 确保检查点之间的间隔
execution.checkpointing.min-pause 60000
// 设置checkpoint的超时时间
execution.checkpointing.timeout 60000
// 设置任务取消后保留hdfs上的checkpoint文件
execution.checkpointing.externalized-checkpoint-retention RETAIN_ON_CANCELLATION

‍

只需要像上面这样就可以配置任务的checkpoint了,当然除了配置这些参数外也可以设置任务的jm,tm等参数,Interpreters 设置里面的很多参数(比如下图的这些)都是可以在note里面直接执行的,然后在这个note里面的任务都会使用刚才的配置.

在执行这些配置的时候需要先把Interpreters 重启一下,因为当Interpreters 进程已启动时就无法更改Interpreters 的属性了,否则会遇到下面的报错

java.io.IOException: Can not change interpreter properties when interpreter process has already been launchedat org.apache.zeppelin.interpreter.InterpreterSetting.setInterpreterGroupProperties(InterpreterSetting.java:957)at org.apache.zeppelin.interpreter.ConfInterpreter.interpret(ConfInterpreter.java:73)at org.apache.zeppelin.notebook.Paragraph.jobRun(Paragraph.java:458)at org.apache.zeppelin.notebook.Paragraph.jobRun(Paragraph.java:72)at org.apache.zeppelin.scheduler.Job.run(Job.java:172)at org.apache.zeppelin.scheduler.AbstractScheduler.runJob(AbstractScheduler.java:130)at org.apache.zeppelin.scheduler.FIFOScheduler.lambda$runJobInScheduler$0(FIFOScheduler.java:39)at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)at java.lang.Thread.run(Thread.java:748)

创建kafak流表

然后来一个非常简单的聚合查询,为了体现任务是从状态中恢复的,所以我们这里根据name分组计算了一个count(age)的操作

这里我先写入了10条数据,jason和spark都是5条,然后先来看一下HDFS上的checkpoint文件是否生成了.

可以看到此时任务已经完成了4次checkpoint了,这时候把任务cancel掉,然后从上一次成功的checkpoint处恢复任务,也就是第4个checkpoint的地方.

%flink.conf
execution.savepoint.path hdfs://master:9000//flink-rockdb/checkpoints/904d58cbd50821f42f33ffed91989ba0/chk-4

需要指定刚才任务的checkpoint路径,先执行上面的语句,然后再接着执行刚才的那条SQL就可以了.

从上面的图可以很清楚的看到,任务是从第4个checkpoint处恢复计算的,然后再来往kafka中写入10条数据看下结果是否正确.

可以看到结果是正确的,是从刚才的状态里面的值接着计算的,说明任务从checkpoint里面恢复成功了.整个过程是不需要写任何代码的,只需要在note里面执行几行配置就可以了,使用起来还是非常方便的.

这篇文章主要是介绍Flink on zeppelin如何设置任务的checkpoint,以及任务重启的时候怎么从指定的checkpoint位置恢复任务.

推荐阅读

Flink on zeppelin第四弹hive streaming writing

JasonLee，公众号：JasonLee的博客Flink on zeppelin第四弹hive streaming writing

Flink on zeppelin第三弹UDF的使用

JasonLee，公众号：JasonLee的博客Flink on zeppelin第三弹UDF的使用

Flink on zeppelin 实时计算pv,uv结果写入mysql

JasonLee，公众号：JasonLee的博客Flink on zeppelin体验第二弹

Flink on zepplien的安装配置

JasonLee，公众号：JasonLee的博客Flink on zepplien的使用体验

更多spark和flink的内容可以关注下面的公众号