Flink版本: 1.8.0
Scala版本: 2.11
Github地址：https://github.com/shirukai/flink-examples-debug-state.git
在本地开发带状态的Flink任务时，经常会遇到这样的问题，需要验证状态是否生效？以及重启应用之后，状态里的数据能否从checkpoint的恢复？首先要明确的是，Flink重启时不会自动加载状态，需要我们手动指定checkpoint路径。笔者从Spark的Structured Streaming转到Flink的时候，就遇到这样的问题。在Spark中，我们使用的状态信息会随着程序再次启动时自动被加载出来。所以当时以为Flink状态也会被自动加载，在开发有状态算子时，测试重启应用之后，并没有继续上一次的状态。一开始以为是checkpoint的设置的问题，调试了好长时间，发现flink需要手动指定checkpoint路径。本篇文章，将从搭建项目到编写带状态的任务，介绍如何在IDEA中调试local模式下带状态的flink任务。
注意：后期git上的项目名称从debug-flink-state-example改为flink-examples-debug-state
Flink提供了Meven模板，能够帮助我们快速创建Maven项目。执行如下命令快速创建一个flink项目：
mvn archetype:generate -DarchetypeGroupId=org.apache.flink -DarchetypeArtifactId=flink-quickstart-scala -DarchetypeVersion=1.8.0 -DgroupId=flink.examples -DartifactId=flink-examples-debug-state -Dversion=1.0 -Dpackage=flink.debug.state.example -DinteractiveMode=false 1项目创建完成后，使用IDEA打开项目。
对pom.xml稍微做一下修改。
纠正一下上面这个问题，flink的两个包作用域都设置为了provided，在程序执行时汇报类不存在的异常。我们可以注释掉scope作用域，也可以在Maven里勾选带有flink依赖的Profiles。
这里我们编写一个简单的Flink任务，实现功能如下
从SocketTextStream中实时接收文本内容将接收到文本转换为事件样例类，事件样例类包含三个字段id、value、time事件按照id进行KeyBy之后，使用process function统计每种事件的个数和value值的总和控制台输出统计结果逻辑比较简单，直接贴代码吧。
package debug.flink.state.example import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor} import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment} import org.apache.flink.api.scala._ import org.apache.flink.configuration.Configuration import org.apache.flink.streaming.api.functions.KeyedProcessFunction import org.apache.flink.util.Collector /** * 实时计算事件总个数，以及value总和 * * @author shirukai */ object EventCounterJob { def main(args: Array[String]): Unit = { // 获取执行环境 val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment // 1. 从socket中接收文本数据 val streamText: DataStream[String] = env.socketTextStream("127.0.0.1", 9000) // 2. 将文本内容按照空格分割转换为事件样例类 val events = streamText.map(s => { val tokens = s.split(" ") Event(tokens(0), tokens(1).toDouble, tokens(2).toLong) }) // 3. 按照时间id分区，然后进行聚合统计 val counterResult = events.keyBy(_.id).process(new EventCounterProcessFunction) // 4. 结果输出到控制台 counterResult.print() env.execute("EventCounterJob") } } /** * 定义事件样例类 * * @param id 事件类型id * @param value 事件值 * @param time 事件时间 */ case class Event(id: String, value: Double, time: Long) /** * 定义事件统计器样例类 * * @param id 事件类型id * @param sum 事件值总和 * @param count 事件个数 */ case class EventCounter(id: String, var sum: Double, var count: Int) /** * 继承KeyedProcessFunction实现事件统计 */ class EventCounterProcessFunction extends KeyedProcessFunction[String, Event, EventCounter] { private var counterState: ValueState[EventCounter] = _ override def open(parameters: Configuration): Unit = { super.open(parameters) // 从flink上下文中获取状态 counterState = getRuntimeContext.getState(new ValueStateDescriptor[EventCounter]("event-counter", classOf[EventCounter])) } override def processElement(i: Event, context: KeyedProcessFunction[String, Event, EventCounter]#Context, collector: Collector[EventCounter]): Unit = { // 从状态中获取统计器，如果统计器不存在给定一个初始值 val counter = Option(counterState.value()).getOrElse(EventCounter(i.id, 0.0, 0)) // 统计聚合 counter.count += 1 counter.sum += i.value // 发送结果到下游 collector.collect(counter) // 保存状态 counterState.update(counter) } }使用nc命令监听9000端口
nl -lk 9000启动flink任务，并模拟如下数据发送
event-1 1 1591695864473 event-1 12 1591695864474 event-2 8 1591695864475 event-1 10 1591695864476 event-2 50 1591695864477 event-1 6 1591695864478效果如下动图所示：
上一步我们已经编写了一个有状态的简单任务，但是状态并没有被持久化，程序重启之后状态会丢失。这时候我们需要给flink任务配置checkpoint。需要简单配置3个地方：
开启checkpoint，并设置做两个checkpoint的间隔设置取消任务时自动保存checkpoint设置基于文件的状态后端 // 配置checkpoint // 做两个checkpoint的间隔为1秒 env.enableCheckpointing(1000) // 表示下 Cancel 时是否需要保留当前的 Checkpoint，默认 Checkpoint 会在整个作业 Cancel 时被删除。Checkpoint 是作业级别的保存点。 env.getCheckpointConfig.enableExternalizedCheckpoints(ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION) // 设置状态后端：MemoryStateBackend、FsStateBackend、RocksDBStateBackend，这里设置基于文件的状态后端 env.setStateBackend(new FsStateBackend("file:///tmp/checkpoints/event-counter"))启动程序，同样模拟数据发送。
这次先发送前三条数据
event-1 1 1591695864473 event-1 12 1591695864474 event-2 8 1591695864475从以上动图中的日志可以看出，flink每隔一秒都会在做checkpoint。
15:59:32,989 INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Triggering checkpoint 102 @ 1592035172989 for job 0c3d201188fc9953cb65498adb4954f4. 15:59:32,997 INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Completed checkpoint 102 for job 0c3d201188fc9953cb65498adb4954f4 (21340 bytes in 7 ms). 15:59:33,990 INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Triggering checkpoint 103 @ 1592035173989 for job 0c3d201188fc9953cb65498adb4954f4. 15:59:34,001 INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Completed checkpoint 103 for job 0c3d201188fc9953cb65498adb4954f4 (21340 bytes in 11 ms). 15:59:34,989 INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Triggering checkpoint 104 @ 1592035174989 for job 0c3d201188fc9953cb65498adb4954f4. 15:59:35,006 INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Completed checkpoint 104 for job 0c3d201188fc9953cb65498adb4954f4 (21340 bytes in 15 ms).查看checkpoint 的目录，发现有checkpoint生成。
ls /tmp/checkpoints/event-counter这里简单说明一下checkpoint目录，程序每次启动都会在指定的目录下（如/tmp/checkpoints/event-counter）根据id生成一个目录，该目录会包含三个目录chk-*、shared、taskowned，每秒做的状态会报存在chk-*目录下，整体目录结构如下所示：
/tmp/checkpoints └── event-counter └── 0c3d201188fc9953cb65498adb4954f4 ├── chk-104 │ ├── 01f2561f-ca48-4699-bbea-40fc849b2b0f │ ├── 021a7b75-f034-4da3-ad0c-e9801a8f1141 │ ├── 17fcf354-c212-43ec-8e7c-99e37a7653c9 │ ├── 33af50a1-e2cb-4364-a723-4c182c5fdb47 │ ├── 3fa88dc7-ea81-4735-83ba-3d4630b7b8ac │ ├── 792068d4-2f89-4d21-aa27-88ef61c7fa99 │ ├── 793d349b-8029-4cb6-b522-22445ec19bae │ ├── _metadata │ ├── acd28b9b-a0cb-4880-9564-9b9fe3c29200 │ ├── c7cbb990-917a-400d-9838-1ac28c92ea10 │ ├── e202ca66-5f9e-4858-bf15-02ca17a4e2b1 │ ├── e7370373-c4be-4c7c-b6df-d959127b31a3 │ └── eb619830-b102-4449-a29c-59d82b6bfbfe ├── shared └── taskowned重启程序之后再发送后三条数据
event-1 10 1591695864476 event-2 50 1591695864477 event-1 6 1591695864478按照预期，当我们发送event-1 10 1591695864476这条数据时，我们得到的结果应该是EventCounter(event-1,11.5,3)，但实际上得到的是EventCounter(event-1,10.0,1)，很明显之前的状态丢失了，原因在文章开头已经说过，这是由于flink并不会自动加载之前的状态，需要我们手动指定checkpoint，如果使用命令行提交任务的话，可以使用-s参数指定savepoint的目录，那么如果在IDEA里开发测试时如何指定呢？下一章会介绍通过魔改源码的方式，实现checkpoint的加载。
首先讲一下思路，当执行env.execute(“EventCounterJob”)时，程序会根据不同的执行环境选择不同的StreamExecutionEnvironment，flink里有两种执行环境：LocalStreamEnvironment和RemoteStreamEnvironment，当我们在IDEA直接运行时，使用的是LocalStreamEnvironment。通过查看RemoteStreamEnvironment的源码可以发现，它最终在构造JobGraph的时候，会将SavepointRestoreSettings的配置通过JobGraph的setSavepointRestoreSettings方法传入到JobGraph中。而在LocalStreamEnvironment中构造的JobGraph没有传入SavepointRestoreSettings的配置，这里我们需要通过修改源码，给JobGraph添加SavepointRestoreSettings配置。
RemoteStreamEnvironment的源码位置：org.apache.flink.streaming.api.environment.RemoteStreamEnvironment。LocalStreamEnvironment的源码位置：org.apache.flink.streaming.api.environment.LocalStreamEnvironment，它的execute()实现源码如下：
public JobExecutionResult execute(String jobName) throws Exception { // transform the streaming program into a JobGraph StreamGraph streamGraph = getStreamGraph(); streamGraph.setJobName(jobName); JobGraph jobGraph = streamGraph.getJobGraph(); jobGraph.setAllowQueuedScheduling(true); Configuration configuration = new Configuration(); configuration.addAll(jobGraph.getJobConfiguration()); configuration.setString(TaskManagerOptions.MANAGED_MEMORY_SIZE, "0"); // add (and override) the settings with what the user defined configuration.addAll(this.configuration); if (!configuration.contains(RestOptions.BIND_PORT)) { configuration.setString(RestOptions.BIND_PORT, "0"); } int numSlotsPerTaskManager = configuration.getInteger(TaskManagerOptions.NUM_TASK_SLOTS, jobGraph.getMaximumParallelism()); MiniClusterConfiguration cfg = new MiniClusterConfiguration.Builder() .setConfiguration(configuration) .setNumSlotsPerTaskManager(numSlotsPerTaskManager) .build(); if (LOG.isInfoEnabled()) { LOG.info("Running job on local embedded Flink mini cluster"); } MiniCluster miniCluster = new MiniCluster(cfg); try { miniCluster.start(); configuration.setInteger(RestOptions.PORT, miniCluster.getRestAddress().get().getPort()); return miniCluster.executeJobBlocking(jobGraph); } finally { transformations.clear(); miniCluster.close(); } }这段代码的大体逻辑是这样的：
获取StreamGraph从StreamGraph中获取JobGraph构造配置创建一个MiniCluster将生成的JobGraph提交给MiniCluster我们可以在提交JobGraph给MiniCluster之前，将SavepointRestoreSettings动态设置给JobGraph，从而实现加载指定savepoint的目的。
上面魔改的代码部分思路是：从Job的全局参数中拿到最后一个checkpoint的路径，这个路径是我们传入进来的。然后通过jobGraph.setSavepointRestoreSettings(SavepointRestoreSettings.forPath(checkpointPath));设置到JobGraph中。
最后，需要修改主程序，让其自动获取最后一个checkpoint路径，然后传入给Job全局参数，添加代码如下：
var params: ParameterTool = ParameterTool.fromArgs(args) val checkPointDirPath = params.get("checkpoint-dir") // 获取最后一个checkpoint文件夹 val checkpointDirs = new io.Directory(new File(checkPointDirPath)).list if (checkpointDirs.nonEmpty) { val lastCheckpointDir = checkpointDirs.maxBy(_.lastModified) val checkpoints = new Directory(lastCheckpointDir.jfile).list.filter(_.name.startsWith("chk-")) if (checkpoints.nonEmpty) { val lastCheckpoint = checkpoints.maxBy(_.lastModified).path val newArgs = Array("--last-checkpoint", "file://" + lastCheckpoint) // 重新载入配置 params = ParameterTool.fromArgs(args ++ newArgs) } } env.getConfig.setGlobalJobParameters(params) // ################################省略代码…… // 设置状态后端：MemoryStateBackend、FsStateBackend、RocksDBStateBackend，这里设置基于文件的状态后端 env.setStateBackend(new FsStateBackend("file://"+checkPointDirPath))测试之前，先清除已有checkpoint
rm -rf /tmp/checkpoints/event-counter命令行执行nc -lk 9000
启动程序，指定参数–checkpoint-dir /tmp/checkpoints/event-counter
先发送三条数据
event-1 1 1591695864473 event-1 12 1591695864474 event-2 8 1591695864475重启应用
再发送三条数据
event-1 1 1591695864473 event-1 12 1591695864474 event-2 8 1591695864475经过魔改后的LocalStreamEnvironment，能够在程序启动时，自动的从指定的checkpoint目录获取最近一次的提交任务的最新的checkpoint，然后指定给JobGraph，使我们的程序能够加载到之前的状态。这种方式只是为了在本地验证状态的可用性，方便我们对状态进行调试，有这种需求的同学，不妨试一下，代码已经提交到github上了，另外有更好的方法，可以一起交流。
