举个例子，当用户停止任务、更新代码逻辑并且改变任务并发度时，两种 state 都是怎样进行恢复的
1. 区分 operator-state 和 keyed-state 的方式
a. operator-state：
1. 状态适用算子：所有算子都可以使用 operator-state，没有限制。
2. 如果需要使用 operator-state，需要实现 CheckpointedFunction 或 ListCheckpointed 接口
3. DataStream API 中，operator-state 提供了 ListState、BroadcastState、UnionListState 3 种用户接口
4. 状态的存储粒度：以单算子单并行度粒度访问、更新状态
5. 并行度变化时：
a. ListState：均匀划分到算子的每个 sub-task 上
b. BroadcastState：每个 sub-task 的广播状态都一样
c. UnionListState：将原来所有元素合并，合并后的数据每个算子都有一份全量状态数据
b. keyed-state：
1. 状态适用算子：keyed-stream 后的算子使用。注意这里很多同学会犯一个错误，就是大家会认为 keyby 后面跟的所有算子都使用的是 keyed-state，但这是错误的 ，比如有 keyby.process.flatmap，其中 flatmap 中使用状态的话是 operator-state
2. 从 context 接口获取具体的 keyed-state
3. DataStream API 中，keyed-state 提供了 ValueState、MapState、ListState 等用户接口，其中最常用 ValueState、MapState
4. 状态的存储粒度：以单 key 粒度访问、更新状态。举例，当我们使用 keyby.process，在 process 中处理逻辑时，其实每一次 process 的处理 context 都会对应到一个 key，所以在 process 中的处理都是以 key 为粒度的。这里很多同学会犯一个错 ，比如想在 open 方法中访问、更新 state，这是不行的，因为 open 方法在执行时，还没有到正式的数据处理环节，上下文中是没有 key 的。
5. 并行度变化时：keyed-state 的重新划分是随着 key-group 进行的。其中 key-group 的个数就是最大并发度的个数。其中一个 key-group 处理一段区间 key 的数据，不同 key-group 处理的 key 是完全不同的。当任务并行度变化时，会将 key-group 重新划分到算子不同的 sub-task 上，任务启动后，任务数据在做 keyby 进行数据 shuffle 时，依然能够按照当前数据的 key 发到下游能够处理这个 key 的 key-group 中进行处理。注意：最大并行度和 key-group 的个数绑定，所以如果想恢复任务 state，最大并行度是不能修改的。大家需要提前预估最大并行度个数
