flink系列-9、flink的狀態與容錯

代碼地址:https://gitee.com/xiexiandong/abc_bigdata.git

1、理解State(狀態)

1.1、State

  • 對象的狀態
    • Flink中的狀態:一般指一個具體的task/operator某時刻在內存中的狀態(例如某屬性的值)
    • 注意:State和Checkpointing 不要搞混 
    • checkpoint則表示了一個Flink Job,在一個特定時刻的一份全狀態快照,即包含一個job下所有task/operator 某時刻的狀態 
  • 狀態的作用
    • 增量計算 
    • 聚合操作 
    • 機?學習訓練模式 
    • 等等 
    • 容錯 
    • Job故障重啟 
    • 升級

1.2、狀態的分類

1、Operator State

  • 綁定到特定operator并行實例,每個operator的并行實例維護一個狀態
  • 與key無關
  • 例如:一個并行度為3的source,如果只考慮一個算子需要一個邏輯狀態的情形,那么他就有3operator個狀態
  • 支持的數據類型
    • ListState

2、Keyed State

  • 基于KeyedStream之上的狀態,dataStream.keyBy(),只能在作用于KeyedStrem上的function/Operator里使用
  • KeyBy之后的Operator State,可理解為分區過的Operator State
  • 每個并行keyed Operator的每個實例的每個key有一個Keyed State:即就是 一個唯一的狀態,由于每個key屬于一個keyed operator的并行實例,因此我們可以將其簡單地理解為 <operator,key>

  

  • 支持的數據結構 
    • ValueState:保留一個可以更新和檢索的值 
      • update(T) 
      • value() 
    • ListState<T>:保存一個元素列表 
      • add(T) 
      • addAll(List) 
      • get(T) 
      • clear() 
    • ReducingState<T>:保存一個值,該值表示添加到該狀態所有值的聚合。 
      • add(T) 
    • AggregatingState<IN,OUT><in,out>:保存一個值,該值表示添加到該狀態的所有值的聚合。(與ReducingState 相反,聚合類型添加到該狀態的元素可以有不同類型) 
      • add(T) 
    • FoldingState<T,ACC><t,acc>:不推薦使用 
      • add(T) 
    • MapState<UK,UV><uk,uv>:保存一個映射列表 
      • put(UK,UV) 
      • putAll(Map<uk,uv>) 
      • get(UK) 

 3、狀態的表現形式

  • Keyed State和Operator State,可以以兩種形式存在:原始狀態和托管狀態。
    • managed(托管狀態):
      • 托管狀態是指Flink框架管理的狀態,如ValueState,ListState,MapState等。 
      • 通過框架提供的接口來更新和管理狀態的值 
      • 不需要序列化 
    • raw(原始狀態) 原始狀態是由用戶自行管理的具體的數據結構,Flink在做checkpoint的時候,使用byte[]來讀寫狀態內 容,對其內部數據結構一無所知
      • 需要序列化
      • 通常在DataStream上的狀態推薦使用托管的狀態,當用戶自定義operator時,會使用到原始狀態。
      • 大多數都是托管狀態,除非自定義實現。

4、Operator State與Keyed State的Redistribute(重新分配)

1)、Operator State Redistribute Redistribute

  • 當Operator改變并發度的時候(Rescale),會觸發狀態的Redistribute,即Operator State里的 數據會重新分配到Operator的Task實例
  • 例如:某Operator的并行度由3改為2

  • 不同數據結構的動態擴展方式不一樣:
    • ListState:并發度在改變的時候,會將并發上的每個List都取出,然后把這些List合并到一個新的List,然 后根據元素的個數在均勻分配給新的Task
    • UnionListState:相比于ListState更加靈活,把劃分的方式交給用戶去做,當改變并發的時候,會將原來 的List拼接起來。然后不做劃分,直接交給用戶(每個Task給全量的狀態,用戶自己劃分)
    • BroadcastState:如大表和小表做Join時,小表可以直接廣播給大表的分區,在每個并發上的數據都是完 全一致的。做的更新也相同,當改變并發的時候,把這些數據COPY到新的Task即可。
    • 以上是Flink Operator States提供的3種擴展方式,用戶可以根據自己的需求做選擇。

2)、Keyed State的Redistribute

  • Keyed State Redistribute
    • Key被Redistribute哪個task,他對應的Keyed State就被Redistribute到哪個Task
    • Keyed State Redistribute是基于Key Group來做分配的:
      • 將key分為group
      • 每個key分配到唯一的group 
      • 將group分配給task實例 
      • KeyGroup由最大并行度的大小所決定的 
      • Keyed State最終分配到哪個Task:group ID和taskID是從0開始算的 
        • hash=hash(key) 
        • KG=hash % numOfKeyGroups 
        • Subtask=KG* taskNum / numOfKeyGroups

二、CheckPoint

 2.1、狀態容錯

  • 有了狀態自然需要狀態容錯,否則狀態就失去意義了
  • Flink狀態容錯的機制就是checkpoint

概念

  • 所謂checkpoint,就是在某一時刻,將所有task的狀態做一個快照(snapshot),然后存儲到State Backend (有全量 和 增量)
  • 一種連續性繪制數據流狀態的機制(周期性的),該機制確保即使出現故障,程序的狀態最終也將為數據流中的每一條記錄提供exactly once(只處理一次)的語意保證(只能保證flink系統內,對于sink和source需要依賴的外部的組件一同保證)
  • 全局快照,持久化保存所有的task / operator的State

特點: 

  • 輕量級容錯機制
  • 可異步
  • 全量 vs 增量
  • Barrier機制(保證exactly-once 語義)
  • 失敗情況可回滾至最近一次成功的checkpoint(自動)
  • 周期性(無需人工干預)

基本原理:

  • 通過往source 注入barrier
  • barrier作為checkpoint的標志
  • barrier
    • 全局異步化是snapshot的核心機制

    • Flink分布式快照的核心概念之一就是數據柵欄(barrier)。這些barrier被插入到數據流中,作為數據流的一部分和數據一起向下流動。Barrier不會干擾正常數據,數據嚴格有序。一個barrier把數據流分割成兩部分:一部 分進入到當前快照,另一部分進入下一個快照。每一個barrier都帶有快照ID,并且barrier之前的數據都進入了 此快照。Barrier不會干擾數據流處理,所以非常輕量。多個不同快照的多個barrier會在流中同時出現,即多個 快照可能同時創建。

使用Checkpointing的前提條件:

  • 在一定時間內可回溯的datasource(故障時可以回溯數據),常見的:
    • 一般是可持久化的消息隊列:例如Kafka、RabbitMQ、Amazon Kinesis、Google PubSub
    • 也可以是文件系統:HDFS、S3、GFS、NFS、Ceph
  • 可持久化存儲State的存儲系統,通常使用分布式文件系統(Checkpointing就是把job的所有狀態都周期性持 久化到存儲里)
    • 一般是HDFS、S3、GFS、NFS、Ceph

2.2、狀態容錯示意圖

checkpoint:

 

Restore:

  • 恢復所有狀態
  • 設置source的位置(例如:Kafka的offset)

 2.3、使用CheckPoint

 1、開啟checkPoint

  • checkPoint默認是禁用的
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)

2、CheckpointConfig設置說明

  • checkpointMode 
    • //set mode to exactly-once (this is the default)
      env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
  • 保留策略 
    • 默認情況下,檢查點不被保留,僅用于從故障中恢復作業??梢詥⒂猛獠砍志没瘷z查點,同時指定保留策略

    • ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION:在作業取消時保留檢查點。注意,在這種情況下,必須在取消后手動清理檢查點狀態。

    • ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION當作業被cancel時,刪除檢查點。檢查點狀態僅在作業失敗時可用。

  • checkpointing的超時時間:超過時間沒有完成則會被終止
    • //checkpoints have to complete within one minute, or are discarded
      env.getCheckpointConfig().setCheckpointTimeout(60000);
  • checkpointing最小間隔:用于指定上一個checkpoint完成之后最小等多久可以出發另一個checkpoint,當指 定這個參數時,maxConcurrentCheckpoints的值為1
    • //make sure 500 ms of progress happen between checkpoints
      env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);
  • maxConcurrentCheckpoints:指定運行中的checkpoint最多可以有多少個(設定checkpointing最小間隔時本 參數即為1) 
    • //allow only one checkpoint to be in progress at the same time
      env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
  • failOnCheckpointingErrors用于指定在checkpoint發生異常的時候,是否應該fail該task,默認為true,如果設 置為false,則task會拒絕checkpoint然后繼續運行
    • env.getCheckpointConfig().setFailOnCheckpointingErrors(true);

3、選擇State Backend 

  • State Backend就是用來保存快照的地方
  • 用來在Checkpointing機制中持久化所有狀態的一致性快照,這些狀態包括:
    • 非用戶定義的狀態:例如,timers、非用戶自定義的stateful operators(connectors,windows)
    • 用戶定義的狀態:就是前面講的用戶自定義的stateful operato所使用的Keyed State and Operator State

目前Flink自帶三個開箱即用State Backend: 

  • MemoryStateBackend(默認)
    • MemoryStateBackend在Java堆上維護狀態。Key/value狀態和窗口運算符使用哈希表存儲值和計時?等
    • Checkpoint時,MemoryStateBackend對State做一次快照,并在向JobManager發送Checkpoint確認完 成的消息中帶上此快照數據,然后快照就會存儲在JobManager的堆內存中 
    • MemoryStateBackend可以使用異步的方式進行快照(默認開啟),推薦使用異步的方式避免阻塞。如果 不希望異步,可以在構造的時候傳入false(也可以通過全局配置文件指定),如下
      • StateBackend backend = new MemoryStateBackend(10*1024*1024,false);
        env.setStateBackend(backend);
    • 限制

      • 單個State的大小默認限制為5MB,可以在MemoryStateBackend的構造函數中增加

      • 不論如何配置,State大小都無法大于akka.framesize(JobManager和TaskManager之間發送的最大消息 的大小默認是10MB)

      • JobManager必須有足夠的內存大小

    •  適用場景 
      • 本地開發和調試 
      • 小狀態job,如只使用Map、FlatMap、Filter...或Kafka Consumer
  • FsStateBackend

    • FsStateBackend需要配置一個文件系統的URL, 如 "hdfs://namenode:40010/flink/checkpoint"。

    • FsStateBackend在TaskManager的內存中持有正在處理的數據。Checkpoint時將state snapshot 寫入文件系 統目錄下的文件中。文件的路徑等元數據會傳遞給JobManager,存在其內存中。

    • FsStateBackend可以使用異步的方式進行快照(默認開啟),推薦使用異步的方式避免阻塞。如果不希望異 步可以在構造的時候傳入false(也可以通過全局配置文件指定),如下:

      • StateBackend backend = new FsStateBackend("hdfs://namenode:40010/flink/checkpoints",false);
        env.setStateBackend(backend);
    • 適用場景 
      • 大狀態、長窗口、大鍵/值狀態的job 
      • 所有高可用性的情況 
  • RocksDBStateBackend

    • RocksDBStateBackend需要配置一個文件系統的URL來, 如"hdfs://namenode:40010/flink/checkpoint"

    • RocksDBStateBackend將運行中的數據保存在RocksDB數據庫中,(默認情況下)存儲在TaskManager數據 目錄中,在Checkpoint時,整個RocksDB數據庫將被Checkpointed到配置的文件系統和目錄中。文件的路徑 等元數據會傳遞給JobManager,存在其內存中。

    • RocksDBStateBackend總是執行異步快照

    • 限制

      • RocksDB JNI API是基于byte[],因此key和value最大支持大小為2^31個字節(2GB)。RocksDB自身在 支持較大value時候有問題

    • 適用場景
      • 超大狀態,超長窗口、大鍵/值狀態的job 
      • 所有高可用性的情況 
    • 與前兩種狀態后端對比: 
      • 目前只有RocksDBStateBackend支持增量checkpoint(默認全量) 
      • 狀態保存在數據庫中,即使用RockDB可以保存的狀態量僅受可用磁盤空間量的限制,相比其他的狀態后 端可保存更大的狀態,但開銷更大(讀/寫需要反序列化/序列化去檢索/存儲狀態),吞吐受到限制 
  • 三種StateBackend總結如下:

  • 配置StateBackend 
    • 全局配置(配置文件conf/flink-conf.yaml),設置集群保存checkpoint類型和存儲路徑
    • # The backend that will be used to store operator state checkpoints
      state.backend: filesystem
      #Directory.for storing checkpoints
      state.checkpoints.dir: hdfs:namenode:40010/flink/checkpoints
    • 每個job單獨配置State Backend(可覆蓋全局配置) ,設置計算任務的
    • StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
      env.setStateBackend(new FsStateBackend("hdfs://namenode:40010/flink/checkpoints"));

4、配置恢復策略

Flink支持不同的重啟策略,這些策略控制在出現故障時如何重新啟動job

  • 如果沒有啟用checkpointing,則使用無重啟(no restart)策略。
  • 如果啟用了checkpointing,但沒有配置重啟策略,則使用固定延遲(fixed-delay)策略,其中嘗試重啟 次數是Integer > MAX_VALUE
  • 重啟策略可以在flink-conf.yaml中配置,表示全局的配置。也可以在應用代碼中動態指定,會覆蓋全局配 置

 2.4、checkpoint demo

1、operatorState的checkPoint容錯案例:

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()}")
  }
}

2、Keyed State容錯實現方法

  • Keyed State之過期超時策略
    • 由于Keyed State太多,所以flink提供了針對Keyed State TTL的設置
    • 任何類型的keyed State都可以設置TTL。如果TTL已配置,且狀態已過期,則將以最佳方式處理
    • 所有State collection都支持條目級別的TTL,即list、map中的條目獨立expire
    • 用法
      • StateTtlConfig ttlConfig = StateTtlConfig
            .newBuilder(Time.seconds(1))
            .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
            .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
            .build();
        ValueStateDescriptor<String> stateDescriptor = new ValueStateDescriptor<>("text state", String.class)
        stateDescriptor.enableTimeToLive(ttlConfig);
    • Refresh策略(默認是OnCreateAndWrite):設置如何更新keyedState的最后訪問時間

      • StateTtlConfig.UpdateType.Disabled - 禁用TTL,永不過期

      • StateTtlConfig.UpdateType.OnCreateAndWrite - 每次寫操作均更新State的最后訪問時間(Create、 Update)

      • StateTtlConfig.UpdateType.OnReadAndWrite - 每次讀寫操作均更新State的最后訪問時間

    • 狀態可見性(默認是NeverReturnExpired):設置是否返回過期的值(過期尚未清理,此時正好被訪問)
      • StateTtlConfig.StateVisibility.NeverReturnExpired - 永不返回過期狀態 
      • StateTtlConfig.StateVisibility.ReturnExpiredlfNotCleanedUp - 可以返回過期但尚未清理的狀態值 
    • TTL time等級
      • setTimeCharacteristic(TimeCharacteristic timeCharacteristic) 
      • 目前只支持ProcessingTime
  • Keyed State之過期狀態清理
    • 清理策略
    • 默認:已經過期的數據被顯示讀取時才會清理(可能會導致狀態越來越大) 
    • FULL_STATE_SCAN_SNAPSHOT:在checkpoint時清理full snapshot中的expired state 
      • CleanupFullSnapshot() 
      • 不適用于在RocksDB state backend上的incremental checkpointing

KeyedState的checkPoint容錯恢復 :

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)
  }
}

三、SavePoint

概念:

  • savepoint可以理解為是一種特殊的checkpoint,savepoint就是指向checkpoint的一個指針,實際上也是 使用通過checkpointing機制創建的streaming job的一致性快照,可以保存數據源的offset、并行操作狀態 也就是流處理過程中的狀態歷史版本。需要手動觸發,而且不會過期,不會被覆蓋,除非手動刪除。正常 情況下的線上環境是不需要設置savepoint的。除非對job或集群做出重大改動的時候, 需要進行測試運 行。
  • 可以從應用在過去的任意做了savepoint的時刻開始繼續消費,具有可以replay的功能

Savepoint由兩部分組成:

  • 數據目錄:穩定存儲上的目錄,里面的二進制文件是streaming job狀態的快照
  • 元數據文件:指向數據目錄中屬于當前Savepoint的數據文件的指針(絕對路徑)

與Checkpoint的區別:

  • Savepoint相當于備份(類比數據庫備份)、Checkpoint相當于recovery log
  • Checkpoint是Flink自動創建的"recovery log"用于故障自動恢復,由Flink創建,不需要用戶交互。用戶 cancel作業時就刪除,除非啟動了保留機制(External Checkpoint)
  • Savepoint由用戶創建,擁有和刪除,保存點在作業終止后仍然存在。

作用:

  • job開發新版本(更改job graph、更改并行度等等),應用重新發布
  • Flink版本的更新
  • 業務遷移,集群需要遷移,不容許數據丟失
posted @ 2020-04-29 16:57  有潔癖的碼農  閱讀(...)  評論(...編輯  收藏
全民捕鱼游戏怎么玩