StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//start a checkpoint every 1000 ms   1000-checkpoint时间间隔
env.enableCheckpointing(1000);
//advanced options: checkpoint保证形式
//set mode to exactly-once (this is the default)
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
//make sure 500 ms of progress happen between checkpoints 两次间隔最小时间，如果上次没有完成会等待完成在执行下一次
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);
//checkpoints have to complete within one minute,or are discarded ；超时时间
env.getCheckpointConfig().setCheckpointTimeout(60000);
//allow only one checkpoint to be in progress at the same time； checkpoint 并行度
env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
//enable externalized checkpoints which are retained after job cancellation；任务结束，checkpoint是否保留
env.getCheckpointConfig().enableExternalizedCheckpoints(ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION)

import java.util.concurrent.TimeUnit
import org.apache.flink.api.common.functions.RichFlatMapFunction
import org.apache.flink.api.common.restartstrategy.RestartStrategies
import org.apache.flink.api.common.state.{ListState, ListStateDescriptor}
import org.apache.flink.api.common.time.Time
import org.apache.flink.api.common.typeinfo.{TypeHint, TypeInformation}
import org.apache.flink.configuration.{ConfigConstants, Configuration}
import org.apache.flink.runtime.state.filesystem.FsStateBackend
import org.apache.flink.runtime.state.{FunctionInitializationContext, FunctionSnapshotContext}
import org.apache.flink.streaming.api.CheckpointingMode
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction
import org.apache.flink.streaming.api.environment.CheckpointConfig
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.util.Collector

import scala.collection.mutable.ListBuffer

/**
  * OperatorState的checkPoint容错恢复
  * 想知道两次事件 xxd 之间，一共发生多少次其他事件，分别是什么事件
  * 事件流：xxd a a a a a f d d xxd ad d s s d xxd…
  * 当事件流中出现字母e时触发容错
  * 输出：
  * (8,a a a a a f d d)
  * (6,ad d s s d)
  */
object OperatorStateRecovery {
  def main(args: Array[String]): Unit = {
    import org.apache.flink.api.scala._
    //生成配置对象
    val config = new Configuration()
    //开启spark-webui
    config.setBoolean(ConfigConstants.LOCAL_START_WEBSERVER, true)
    //配置webui的日志文件，否则打印日志到控制台
    config.setString("web.log.path", "/tmp/logs/flink_log")
    //配置taskManager的日志文件，否则打印日志到控制台
    config.setString(ConfigConstants.TASK_MANAGER_LOG_PATH_KEY, "/tmp/logs/flink_log")
    //配置tm有多少个slot
    config.setString("taskmanager.numberOfTaskSlots", "4")
    // 获取local运行环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(config)
    //设置全局并行度为1，好让所有数据都跑到一个task中，以方便测试
    env.setParallelism(1)
    //隔多长时间执行一次ck  毫秒
    env.enableCheckpointing(1000L)
    val checkpointConfig: CheckpointConfig = env.getCheckpointConfig
    //保存EXACTLY_ONCE
    checkpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)
    //每次ck之间的间隔，不会重叠
    checkpointConfig.setMinPauseBetweenCheckpoints(2000L)
    //每次ck的超时时间
    checkpointConfig.setCheckpointTimeout(10L)
    //如果ck执行失败，程序是否停止
    checkpointConfig.setFailOnCheckpointingErrors(true)
    //job在执行CANCE的时候是否删除ck数据
    checkpointConfig.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION)
    //指定保存ck的存储模式
    val stateBackend = new FsStateBackend("file:/tmp/flink/checkpoints", true) //异步同步
    // val stateBackend = new MemoryStateBackend(10 * 1024 * 1024,false)
    // val stateBackend = new RocksDBStateBackend("hdfs://ns1/flink/checkpoints",true)
    env.setStateBackend(stateBackend)
    //恢复策略，恢复三次，间隔0秒
    env.setRestartStrategy(
      RestartStrategies.fixedDelayRestart(
        3, // number of restart attempts
        Time.of(0, TimeUnit.SECONDS) // delay
      )
    )
    val input: DataStream[String] = env.socketTextStream("localhost", 6666)
    input
      .flatMap(new OperatorStateRecoveryRichFunction)
      .print()
    env.execute()
  }
}

//由于使用了本地状态所以需要checkpoint的snapshotState方法把本地状态放到托管状态中
class OperatorStateRecoveryRichFunction extends RichFlatMapFunction[String, (Int, String)] with CheckpointedFunction {

  //托管状态
  @transient 
  private var checkPointCountList: ListState[String] = _
  //原始状态
  private var list: ListBuffer[String] = new ListBuffer[String]

  //flatMap函数处理逻辑
  override def flatMap(value: String, out: Collector[(Int, String)]): Unit = {
    if (value == "xxd") {
      if (list.size > 0) {
        val outString: String = list.foldLeft("")(_ + " " + _)
        out.collect((list.size, outString))
        list.clear()
      }
    } else if (value == "e") {
      1 / 0
    } else {
      list += value
    }
  }

  //再checkpoint时存储，把正在处理的原始状态的数据保存到托管状态中
  override def snapshotState(context: FunctionSnapshotContext): Unit = {
    checkPointCountList.clear()
    list.foreach(f => checkPointCountList.add(f))
    println(s"snapshotState:${
      list
    }, Time=${System.currentTimeMillis()}")
  }

  //从statebackend中恢复保存的托管状态，并将来数据放到程序处理的原始状态中
  // 出错一次就调用一次这里，能调用几次是根据setRestartStrategy设置的
  override def initializeState(context: FunctionInitializationContext): Unit = {
    val lsd: ListStateDescriptor[String] = new ListStateDescriptor[String]("xxdListState", TypeInformation.of(new TypeHint[String] {}))
    checkPointCountList = context.getOperatorStateStore.getListState(lsd)
    if (context.isRestored) {// 出错恢复
      import scala.collection.convert.wrapAll._
      for (e <- checkPointCountList.get()) {
        list += e
      }
    }
    println(s"initializeState:${list},Time=${System.currentTimeMillis()}")
  }
}

import java.util.concurrent.TimeUnit

import org.apache.flink.api.common.functions.RichFlatMapFunction
import org.apache.flink.api.common.restartstrategy.RestartStrategies
import org.apache.flink.api.common.state.{StateTtlConfig, ValueState, ValueStateDescriptor}
import org.apache.flink.api.common.time.Time
import org.apache.flink.api.common.typeinfo.{TypeHint, TypeInformation}
import org.apache.flink.configuration.{ConfigConstants, Configuration}
import org.apache.flink.runtime.state.filesystem.FsStateBackend
import org.apache.flink.streaming.api.CheckpointingMode
import org.apache.flink.streaming.api.environment.CheckpointConfig
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.util.Collector

import scala.collection.mutable


/**
  * KeyedState的checkPoint容错恢复
  * 将输入格式为"字符串 数字"的字符串转换成(字符串,数字)的元组类型
  * 事件流：xxd 666
  * 当事件流中出现"任意字符串 888"时触发容错
  * 输出：
  * (xxd,666)
  */
object KeyedStateRecovery {
  def main(args: Array[String]): Unit = {
    import org.apache.flink.api.scala._
    //生成配置对象
    val config = new Configuration()
    //开启spark-webui
    config.setBoolean(ConfigConstants.LOCAL_START_WEBSERVER, true)
    //配置webui的日志文件，否则打印日志到控制台
    config.setString("web.log.path", "/tmp/logs/flink_log")
    //配置taskManager的日志文件，否则打印日志到控制台
    config.setString(ConfigConstants.TASK_MANAGER_LOG_PATH_KEY, "/tmp/logs/flink_log")
    //配置tm有多少个slot
    config.setString("taskmanager.numberOfTaskSlots", "4")
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(config)
    //并行度设置为1，是想让所有的key都跑到一个task中，以方便测试
    env.setParallelism(1)
    //隔多长时间执行一次ck
    env.enableCheckpointing(1000L)
    val checkpointConfig: CheckpointConfig = env.getCheckpointConfig
    //保存EXACTLY_ONCE
    checkpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)
    //每次ck之间的间隔，不会重叠
    checkpointConfig.setMinPauseBetweenCheckpoints(2000L)
    //每次ck的超时时间
    checkpointConfig.setCheckpointTimeout(10L)
    //如果ck执行失败，程序是否停止
    checkpointConfig.setFailOnCheckpointingErrors(true)
    //job在执行CANCE的时候是否删除ck数据
    checkpointConfig.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION)
    //指定保存ck的存储模式
    val stateBackend = new FsStateBackend("file:/tmp/flink/checkpoints", true)
    // val stateBackend = new MemoryStateBackend(10 * 1024 * 1024,false)
    // val stateBackend = new RocksDBStateBackend("hdfs://ns1/flink/checkpoints",true)
    env.setStateBackend(stateBackend)
    //恢复策略
    env.setRestartStrategy(
      RestartStrategies.fixedDelayRestart(
        3, // number of restart attempts
        Time.of(3, TimeUnit.SECONDS) // delay
      )
    )
    val input: DataStream[String] = env.socketTextStream("localhost", 6666)
    //因为KeyedStateRichFunctionString中使用了keyState，所以它必须在keyBy算子的后面
    input
      .map(f => {
        val strings: mutable.ArrayOps[String] = f.split(" ")
        (strings(0), strings(1).toInt)
      })
      .keyBy(0)
      .flatMap(new KeyedStateRecoveryRichFunctionString)
      .print()
    env.execute()
  }
}

//由于没有使用本地的状态所以不需要实现checkpoint接口
class KeyedStateRecoveryRichFunctionString extends RichFlatMapFunction[(String, Int), (String, Int)] {
  //ValueState是Key的state类型，是只能存在于KeyedStream的operator中
  @transient private var sum: ValueState[(String, Int)] = null

  override def flatMap(value: (String, Int), out: Collector[(String, Int)]): Unit = {
    println(s"state value:${sum.value()}")
    //当value值为888时，触发异常
    if (value._2 != 888) {
      sum.clear()
      sum.update(value)
      out.collect(value)
    } else {
      1 / 0
    }
  }

  //在operator启动时执行一次
  //如果operator出现异常，在恢复operator时会被再次执行
  override def open(parameters: Configuration): Unit = {
    //keyState的TTL策略
    val ttlConfig = StateTtlConfig
      //keyState的超时时间为10秒
      .newBuilder(Time.seconds(10))
      //当创建和更新时，重新计时超时时间
      .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
      //失败时不返回keyState的值
      .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
      //失败时返回keyState的值
//      .setStateVisibility(StateTtlConfig.StateVisibility.ReturnExpiredIfNotCleanedUp)
      //ttl的时间处理等级目前只支持ProcessingTime
      .setTtlTimeCharacteristic(StateTtlConfig.TtlTimeCharacteristic.ProcessingTime)
      .build
    //从runtimeContext中获得ck时保存的状态
    val descriptor = new ValueStateDescriptor[(String, Int)]("xxdValueState", TypeInformation.of(new TypeHint[(String, Int)] {}))
    descriptor.enableTimeToLive(ttlConfig)
    sum = getRuntimeContext.getState(descriptor)
  }
}