# Flink Savepoints CLI介绍
#### 概述
Savepoints在持久化存储中保存某个checkpoint，以便用户可以暂停自己的应用进行升级，并将状态设置为savepoint的状态，并继续运行。该机制利用了Flink的checkpoint机制创建流应用的快照，并将快照的元数据（meta-data）写入到一个额外的持久化文件系统中。
如果需要使用savepoints的功能，强烈推荐用户为每个算子通过uid(String)分配一个固定的ID，以便将来升级恢复使用，示例代码如下：
```
DataStream<String> stream = env
// Stateful source (e.g. Kafka) with ID
.addSource(new StatefulSource())
.uid("source-id") // ID for the source operator
.shuffle()
 // Stateful mapper with ID
.map(new StateFulMapper())
.uid("mapper-id") // ID for the mapper
// Stateless printing sink
.print(); //Auto-generated ID
```
#### savepoint恢复
如果用户不手动设置ID，系统将自动给每个算子分配一个ID。只要该算子的ID不改变，即可从savepoint恢复，ID的产生取决于用户的应用代码，并且对应用代码的结构十分敏感。因此，强烈推荐用户手动为每个算子设置ID。Savepoint产生的数据将被保存到配置的文件系统中，如FsStateBackend或者RocksDBStateBackend。
1. 触发一个savepoint
   ```
   $ bin/flink savepoint <jobId> [targetDirectory]
   ```
   以上命令将触发ID为jobId的作业产生一个savepoint，另外，用户可以通过targetDirectory指定savepoint的存储路径，该路径必须是jobManager可以访问的，由于targetDirectory是可选的，如果用户没有配置targetDirectory，则是使用配置文件中"state.savepoints.dir"配置的目录来存放savepoint。
   用户可以在"flink-conf.yaml"中通过"state.savepoints.dir"选项设置默认的savepoint路径。
   ```
   # Default savepoint target directory
   ```
   ![](https://support.huaweicloud.com/devg3-mrs/public_sys-resources/note_3.0-zh-cn.png)
   建议用户将targetDirectory路径设置为HDFS路径 ，例如：
   ```
   bin/flink savepoint  405af8c02cf6dc069a0f9b7a1f7be088 hdfs://savepoint
   ```
   
2. 删除一个作业并进行savepoint
   ```
   $ bin/flink cancel -s [targetDirectory] jobId
   ```
   以上命令将删除一个作业，同时，在删除前将对该作业的状态进行保存。另外，用户可以通过targetDirectory指定savepoint的存储路径，该路径必须是jobManager可以访问的。
   
3. 恢复作业方式
   - 从savepoint恢复作业。
     ```
     $ bin/flink run -s savepointPath [runArgs]
     ```
     以上命令将提交一个作业，并将该作业的初始状态置为savepointPath指定的状态。
     ![](https://support.huaweicloud.com/devg3-mrs/public_sys-resources/note_3.0-zh-cn.png)
     runArgs是指用户应用中自定义的参数，每个用户自定义的参数形式、名称都不一样。
     
   
   - 允许不恢复某个算子的状态
     ```
     $ bin/flink run -s savepointPath -n [runArgs]
     ```
     默认情况下，系统将尝试将savepoint的状态全部映射到用户的流应用中，如果用户升级的流应用删除了某个算子，可以通过--allowNonRestoredState(简写-n)恢复状态。
     
    
4. 清除savepoints
   ```
   $ bin/flink savepoint -d savepointPath
   ```
   以上命令将删除保存在savepointPath的savepoint。
   
 
#### 注意事项
- 如果一个task中有算子链（Chained operators），将会将算子链上第一个算子的ID分配给该task。给算子链上的中间算子手动分配ID是不可能的。例如：在链（Chain）\[a-\>b-\>c\]中，只能给a手动分配ID，b和c不能分配。如果用户想给b和c分配ID，用户必须手动建链。手动建链时需要使用disableChaining()接口。举例如下：
  ```
  env.addSource(new GetDataSource())
  .keyBy(0)
  .timeWindow(Time.seconds(2)).uid("window-id")
  .reduce(_+_).uid("reduce-id")
  .map(f=>(f,1)).disableChaining().uid("map-id")
  .print().disableChaining().uid("print-id")
  ```
  
- 用户升级job时不允许更改算子的数据类型。