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)

  }

}