Flink里面有state的概念
State:一般指一个具体的task/operator的状态。State可以被记录，在失败的情况下数据还可以恢复，Flink中有两种基本类型的State：Keyed State，Operator State，他们两种都可以以两种形式存在：原始状态(raw state)和托管状态(managed state)
托管状态：由Flink框架管理的状态，我们通常使用的就是这种。
原始状态：由用户自行管理状态具体的数据结构，框架在做checkpoint的时候，使用byte[]来读写状态内容，对其内部数据结构一无所知。通常在DataStream上的状态推荐使用托管的状态，当实现一个用户自定义的operator时，会使用到原始状态。但是我们工作中一般不常用，所以我们不考虑他。
里面没有shuffle的操作，或者说里面没有key by的操作。
（1）operator state是task级别的state，说白了就是每个task对应一个state
（2）Kafka Connector source中的每个分区（task）都需要记录消费的topic的partition和offset等信息。
（3）operator state 只有一种托管状态：ValueState
（4）operator state可以定义为source和sink的状态进行管理
（1）keyed state记录的是每个key的状态
（2）Keyed state托管状态有六种类型
Flink支持的StateBackend:
MemoryStateBackend
FsStateBackend
RocksDBStateBackend
默认情况下，状态信息是存储在 TaskManager 的堆内存中的，checkpoint 的时候将状态保存到JobManager 的堆内存中。
缺点：
只能保存数据量小的状态
状态数据有可能会丢失
优点：
开发测试很方便
状态信息存储在TaskManager 的堆内存中的，checkpoint 的时候将状态保存到指定的文件中 (HDFS等文件系统)
状态大小受TaskManager内存限制(默认支持5M)
状态访问速度很快
状态信息不会丢失
用于： 生产，也可存储状态数据量大的情况
状态信息存储在 RocksDB 数据库 (key-value 的数据存储服务)， 最终保存在本地文件中
checkpoint 的时候将状态保存到指定的文件中 (HDFS 等文件系统)
状态访问速度有所下降
可以存储超大量的状态信息
用于： 生产，可以存储超大量的状态信息
<code>修改当前任务代码 env.setStateBackend(new FsStateBackend("hdfs://namenode:9000/flink/checkpoints")); 或者new MemoryStateBackend() 或者new RocksDBStateBackend(filebackend, true);【需要添加第三方依赖】</code>
<code>修改flink-conf.yaml state.backend: filesystem state.checkpoints.dir: hdfs://namenode:9000/flink/checkpoints 注意：state.backend的值可以是下面几种：jobmanager(MemoryStateBackend), filesystem(FsStateBackend), rocksdb(RocksDBStateBackend)</code>
（1）为了保证state的容错性，Flink需要对state进行checkpoint。 （2）Checkpoint是Flink实现容错机制最核心的功能，它能够根据配置周期性地基于Stream中各个
Operator/task的状态来生成快照，从而将这些状态数据定期持久化存储下来，当Flink程序一旦意外崩
溃时，重新运行程序时可以有选择地从这些快照进行恢复，从而修正因为故障带来的程序数据异常
（3）Flink的checkpoint机制可以与(stream和state)的持久化存储交互的前提：
持久化的source，它需要支持在一定时间内重放事件。这种sources的典型例子是持久化的消息队列
（比如Apache Kafka，RabbitMQ等）或文件系统（比如HDFS，S3，GFS等）
用于state的持久化存储，例如分布式文件系统（比如HDFS，S3，GFS等）
生成快照：5秒
恢复快照：
默认checkpoint功能是disabled的，想要使用的时候需要先启用，checkpoint开启之后，
checkPointMode有两种，Exactly-once和At-least-once，默认的checkPointMode是Exactly-once，
Exactly-once对于大多数应用来说是最合适的。At-least-once可能用在某些延迟超低的应用程序（始终延迟为几毫秒）。
Flink支持不同的重启策略，以在故障发生时控制作业如何重启，集群在启动时会伴随一个默认的重启策略，在没有定义具体重启策略时会使用该默认策略。 如果在工作提交时指定了一个重启策略，该策略会覆盖集群的默认策略，默认的重启策略可以通过 Flink 的配置文件 flink-conf.yaml 指定。配置参数restart-strategy 定义了哪个策略被使用。
常用的重启策略
（1.1）固定间隔 (Fixed delay)
（1.2）失败率 (Failure rate)
（1.3）无重启 (No restart)
如果没有启用 checkpointing，则使用无重启 (no restart) 策略。
如果启用了 checkpointing，但没有配置重启策略，则使用固定间隔 (fixed-delay) 策略， 尝试重启次数默认值是：Integer.MAX_VALUE，重启策略可以在flink-conf.yaml中配置，表示全局的配置。也可以在应用代码中动态指定，会覆盖全局配置。
（2.1）固定间隔 (Fixed delay)
（2.2）失败率 (Failure rate)
（2..3）无重启 (No restart)
默认情况下，如果设置了Checkpoint选项，则Flink只保留最近成功生成的1个Checkpoint，而当Flink程序失败时，可以从最近的这个Checkpoint来进行恢复。但是，如果我们希望保留多个Checkpoint，并能够根据实际需要选择其中一个进行恢复，这样会更加灵活，比如，我们发现最近4个小时数据记录处理有问题，希望将整个状态还原到4小时之前Flink可以支持保留多个Checkpoint，需要在Flink的配置文件conf/flink-conf.yaml中，添加如下配置，指定最多需要保存Checkpoint的个数：
这样设置以后就查看对应的Checkpoint在HDFS上存储的文件目录
<code>hdfs dfs -ls hdfs://namenode:9000/flink/checkpoints</code>
如果希望回退到某个Checkpoint点，只需要指定对应的某个Checkpoint路径即可实现
如果Flink程序异常失败，或者最近一段时间内数据处理错误，我们可以将程序从某一个Checkpoint点进行恢复