Flink（四）狀態管理 1-有解無憂

一、狀態是什么？

1.1有狀態和無狀態：

Flink不是要做流處理嘛，那當一個資料流過來的時候，第一個資料首先會被flink中的算子執行，執行完成后會生成一個執行結果
這個執行后的結果，例如是輸出一下，后續再過來的資料，例如第二個資料的計算就和第一個資料的計算毫無關系了，這就是無狀態
這個執行后的結果，例如是需要做求和的計算，后續再過來的資料例如，的第二個資料的計算需要依賴與第一次計算的結果，這就是有狀態的計算

1.2.Flink中的有狀態流計算

1.2.1 flink中主要使用它的有狀態計算
我們使用Flink這個技術主要也是來完成我們這些有狀態的計算的功能和任務的，因為在流處理的程序中，很多場景下都需要使用前邊一些資料的執行結果作為依賴進行后續的計算
1.2.2 具體的一些場景

篩選資料源過來的資料流中的一條條資料，看看是否是我們想要的格式，如果是就保存起來，如果不是就過濾掉
對一段時間內從資料源傳過來的資訊進行聚合分析，例如說統計10分鐘內傳過來的資料中大于80小于90之間的資料
對從資料源輸入過來的資料線進行記錄，然后在進行去重等操作

上述的例子中都有個特點，就是將之前輸入過來的資料進行保存，然后在對后續過來的資料進行更新，這便是flink中的有狀態計算

1.3 Flink中狀態的更新流程

狀態的更新，說白了就是前面過來的資料的計算結果的更新，可以參考下面這個圖來理解：

在這里插入圖片描述

第一步資料從資料源的輸入：receuve input
拿到了輸入過來的資料后就要在Task中進行計算
然后就到了第二步，get state 得到了狀態，也就是得到了上次計算完之后的結果，如果是第一次的話，就拿到了也是個空的
然后將這個結果保存起來,就是將計算結果更新到state中，既是第三步：update state
然后后續資料就又過來了，也是先在Task中計算，這時候這個第二條以及之后過來的資料他們都要依賴于上一次計算的結果了，這時候執行的第二步就可以從拿到state，也就是拿到上次的計算的結果，然后在進行Task計算，計算完成后將計算的結果更新到state中，也就是途中的第三步：update state

二、狀態分為哪些？

2.1兩種分類方式：

按照類可以劃分為Operator State算子狀態和Keyed State鍵控狀態
按照儲存可以劃分為：托管狀態（Managed State）和原生狀態（Raw State）

2.2 對于Operator State算子狀態和Keyed State鍵控狀態的進一步理解

1.首先說說這兩種狀態
每個狀態（計算結果的資料）都是有當前任務完成的，自然也和當前算子關聯在一起，那 Flink需要對這些狀態進行管理首先就得知道這些狀態定義的型別是什么型別吧，所以一開始就得注冊相應的狀態，也就是所謂的描述器，即Operator State算子狀態和Keyed State鍵控狀態
2.它們兩個的區別：主要區別就是作用范圍不一樣

算子狀態的作用范圍就是限定為算子任務（也就是當前一個磁區執行的時候，所有資料來了都能訪問到狀態）
鍵控狀態中并不是當前磁區所有的資料都能訪問所有的狀態，而是按照keyby之后的key做劃分，當前key只能訪問自己的狀態

2.3 對于托管狀態（Managed State）和原生狀態（Raw State）的進一步理解

Keyed State以及Operator State都會以兩種方式存盤：managed和raw，

managed state指的是由Flink控制state的資料結構，比如使用內部hash表、RocksDB等，正是基于此，Flink可以更好地在managed state基礎上進行記憶體優化和故障恢復，
raw state指的是Flink只知道state是一些位元組陣列，其余一無所知，需要用戶自己完成state的序列化以及反序列化，因此，Flink不能基于raw state進行記憶體優化以及故障恢復，所以在企業實戰中，很少使用raw state

兩者的區別：

Managed State是由Flink管理的，Flink幫忙存盤、恢復和優化，
Raw State是開發者自己管理的，需要自己序列化，

具體區別：

從狀態管理的方式上來說，Managed State由Flink Runtime托管，狀態是自動存盤、自動恢復的，Flink在存盤管理和持久化上做了一些優化，當橫向伸縮，或者說修改Flink應用的并行度時，狀態也能自動重新分布到多個并行實體上，Raw State是用戶自定義的狀態，
從狀態的資料結構上來說，Managed State支持了一系列常見的資料結構，如ValueState、ListState、MapState等，Raw State只支持位元組，任何上層資料結構需要序列化為位元組陣列，使用時，需要用戶自己序列化，以非常底層的位元組陣列形式存盤，Flink并不知道存盤的是什么樣的資料結構，
從具體使用場景來說，絕大多數的算子都可以通過繼承Rich函式類或其他提供好的介面類，在里面使用Managed State，Raw State是在已有算子和Managed State不夠用時，用戶自定義算子時使用，

為了自定義Flink的算子，可以重寫Rich Function介面類，比如RichFlatMapFunction，使用Keyed State時，通過重寫Rich Function介面類，在里面創建和訪問狀態，對于Operator State，還需進一步實作CheckpointedFunction介面，

三、深入了解 Keyed State

3.1 簡介

鍵控鍵控，鍵控狀態肯定是相對于鍵來進行管理、維護和訪問的
Keyed State只能在KeyedStream后使用，Keyed State很類似于一個分布式的key-value map資料結構，只能用于KeyedStream（keyBy算子處理之后），
鍵控狀態基于每個key去管理，一般keyby進行HashCode重磁區后基于它自己獨享的記憶體空間就會針對每一個不同的key分別保存一份獨立的存盤狀態，而且接下來來了一個新的資料只能訪問自己的狀態，不能訪問其他key的，Flink會為每一個key維護一個狀態，
每個狀態都有clear()是清空操作，
在進行狀態編程時需要通過RuntimeContext注冊StateDescriptor，

如圖：就是說有一堆資料過來了，首先會根據key進行分組，分到哪就去執行哪的算子，比如說算子任務2，過來了一條資料，若是被分配到了算子任務二，你就呆在這里面，你所能訪問的狀態（以前計算的結果資料）也就只能是你們這一個算子任務的，若想訪問算子任務一中的狀態，對不起，不能

在這里插入圖片描述

3.2 Flink 提供了以下資料格式來管理和存盤鍵控狀態 (Keyed State)：

ValueState：存盤單值型別的狀態，可以使用 update(T) 進行更新，并通過 T value() 進行檢索，
ListState：存盤串列型別的狀態，可以使用 add(T) 或 addAll(List) 添加元素；并通過 get() 獲得整個串列，
ReducingState：用于存盤經過 ReduceFunction 計算后的結果，使用 add(T) 增加元素，
AggregatingState：用于存盤經過 AggregatingState 計算后的結果，使用 add(IN) 添加元素，
MapState：維護 Map 型別的狀態，

小結如圖：
在這里插入圖片描述

3.3接下來就是在 Flink 中怎么使用這些東西了

大致步驟如下：

通過dataStream.keyBy()獲取到一個keyedStream
keyedStream.map(自己寫的方法)
定義這個自己寫的方法：基于狀態完成資料的處理

3.4 然后主要內容就是如何根據需求不同怎么撰寫這個自定義類

大致步驟：

寫一個類，繼承RichMapFunction類
重寫RichMapFunction里面的open方法，在open方法中，通過RuntimeContext物件的getXxxState(XxxStateDescriptor)方法獲取到XxxState物件
實作RichMapFunction里面的map方法，在map方法中，通過XxxState物件根據業務需要實作具體功能
在代碼中的KeyedStream上使用自定義的MapFunction

3.4.1 需求：wordcount ，練習ValueState

import org.apache.flink.api.common.functions.{RichMapFunction, RuntimeContext}
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.api.java.tuple.Tuple
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala._

/**
 * 通過wordcount功能，看ValueState的應用
 */
object ValueStateJob {
  def main(args: Array[String]): Unit = {
      //1.定義環境
    val environment: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
      //2.設定資料源
    val dataStream: DataStream[String] = environment.socketTextStream("192.168.229.10", 9999)
      //3.進行keyBy 得到keyedStream
    val keyedStream: KeyedStream[(String, Int), Tuple] = dataStream
      .flatMap(_.split(" "))
      .map((_, 1))
      .keyBy(0)

      //4.通過自定義方法實作狀態的具體操作，完成資料的處理
    val result: DataStream[String] = keyedStream.map(new MyMapFunction)
      //5，列印輸出
    result.print()
      //6.執行
    environment.execute("ValueStateJob")
  }

}
//兩個型別引數，分別表示的是輸入型別和輸出型別
//輸入型別：就是使用這個函式的keyedStream中的資料型別
//輸出型別：是根據業務需要自己設定的型別
class MyMapFunction extends RichMapFunction[(String, Int),String]{
  //valueState中存盤的是單詞的個數
  var valueState:ValueState[Int]=_
    
  //open方法，用來做初始化的方法：只執行一次
  //在這個方法里面創建需要的狀態物件
  override def open(parameters: Configuration): Unit = {
    //要創建狀態物件，只需要通過RuntimeContext物件，提供的方法就可以把物件創建出來
    val runtimeContext: RuntimeContext = getRuntimeContext//通過RichMapFunction里面提供的方法getRuntimeContext可以獲取到一個RuntimeContext物件
    //valueStateDescriptor:就是valueState的一個描述者，就是在這個里面宣告ValueState中存盤的資料的型別
    //兩個引數分別表示：唯一標記以及狀態中需要存盤的資料的型別資訊
    var valueStateDescriptor:ValueStateDescriptor[Int]=new ValueStateDescriptor[Int]("valueState",createTypeInformation[Int])
    valueState=runtimeContext.getState(valueStateDescriptor)//通過runtimeContext提供的getState方法可以獲取一個ValueState物件

  }

  //value:就是輸入（流）進來的資料;每流入進來一個元素都會執行一次這個方法
  override def map(value: (String, Int)): String = {
    //在這個方法中完成word count的計算
    //思路：首先從狀態中把word對應的count獲取到，然后加1,加完之后，再把最新的結果存入到狀態中
    //1.通過valueState的value方法，獲取到狀態中存盤的資料
    val oldCount: Int = valueState.value()
    //讓原來的資料加1
    val newCount: Int = oldCount + value._2//也可以這樣寫：oldCount+1
    //2.通過valueState的update方法，把新計算的結果存入到狀態中
    valueState.update(newCount)
    value._1+"==的數量是==>"+newCount
  }
}

3.4.2 需求：實作用戶瀏覽商品類別統計，練習ListState

import java.lang
import org.apache.flink.api.common.functions.RichMapFunction
import org.apache.flink.api.common.state.{ListState, ListStateDescriptor}
import org.apache.flink.api.java.tuple.Tuple
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala._

import scala.collection.JavaConverters._

/**
 * 通過用戶訪問的類別
 * 業務系統發送過來的日志資訊是這樣的格式：用戶編號  用戶名  訪問的類別名
 * 通過狀態完成統計處理
 * 應該根據用戶做統計（keyBy(用戶)）；一個用戶有可能會訪問很多類別：應該使用ListState存盤用戶訪問過的類別
 */
object ListStateJob {
  def main(args: Array[String]): Unit = {
    val environment: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    //模擬采集業務系統的日志資訊；接下來測驗的時候，就應該按照這種格式輸入資料：用戶編號 用戶名 訪問的類別
    val dataStream: DataStream[String] = environment.socketTextStream("192.168.229.10", 9999)
    val keyedStream: KeyedStream[(String, String, String), Tuple] = dataStream
      .map(_.split("\\s+"))
      .map(array => (array(0), array(1), array(2)))
      .keyBy(0)
    val result: DataStream[(String, String)] = keyedStream.map(new MyListStateMapFunction)
    result.print()
    environment.execute("ListStateJob")
  }
}
class MyListStateMapFunction extends RichMapFunction[(String, String, String),(String,String)]{
  var listState:ListState[String]=_
  override def open(parameters: Configuration): Unit = {
    listState=getRuntimeContext.getListState(new ListStateDescriptor[String]("lsd",createTypeInformation[String]))
  }

  override def map(value: (String, String, String)): (String, String) = {
    /*//根據業務需要，從狀態中獲取資料，然后處理資料，然后把資料在保存到狀態中
    listState.add(value._3)//add方法就是往狀態中添加一個資料
    //構建回傳值
    //get方法，獲取到狀態中存盤的資料
    val iter: lang.Iterable[String] = listState.get()
    val scalaIterable: Iterable[String] = iter.asScala//把java的Iterable轉換成scala的Iterable
    val str: String = scalaIterable.mkString(",")//通過mkString方法，把iterable物件中的元素都通過逗號連接起來*/

    //考慮到去重：存盤的資料就是已經去重的資料
    //1.從狀態中資料獲取到，把新進來的資料添加上，然后去重；然后再存入狀態中
    val oldIterable: lang.Iterable[String] = listState.get()
    val scalaList: List[String] = oldIterable.asScala.toList
//    println(scalaList)
    val list: List[String] = scalaList :+ value._3//追加:
//    println(scalaList+"=========================")
    val distinctList: List[String] = list.distinct//去重
    listState.update(distinctList.asJava)//更新狀態中的資料;upate方法需要一個util.list;所以應該通過asJava轉換一下
    (value._1+":"+value._2,distinctList.mkString(" | "))
  }
}

3.4.3 需求：統計用戶瀏覽商品類別以及該類別的次數，練習MapState

import org.apache.flink.api.common.functions.RichMapFunction
import org.apache.flink.api.common.state.{ListState, ListStateDescriptor, MapState, MapStateDescriptor}
import org.apache.flink.api.java.tuple.Tuple
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala._

import scala.collection.JavaConverters._
/**
 * 通過MapState記錄用戶瀏覽的類別以及該類別對應的瀏覽次數
 */
object MapStateJob {
  def main(args: Array[String]): Unit = {
    /**
     * 1.2.3.4.5
     */
    val environment: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    //資料===》用戶編號 用戶名 所訪問的類別
    val dataStream: DataStream[String] = environment.socketTextStream("192.168.229.10", 9999)
    //要處理，就應該根據用戶分組===》根據用戶做keyby
    val keyedStream: KeyedStream[(String, String), Tuple] = dataStream.map(_.split("\\s+"))
      .map(words => (words(0) + ":" + words(1), words(2)))
      .keyBy(0)
    val result: DataStream[String] = keyedStream.map(new MyMapMapFunction)
    result.print()
    environment.execute("MapStateJob")
  }
}
class MyMapMapFunction extends RichMapFunction[(String,String),String]{
  //通過MapState把用戶訪問的類別存盤起來
  //mapState中的key是類別，value是該類別對應的訪問次數
  var mapState:MapState[String,Int]=_
  override def open(parameters: Configuration): Unit = {
    mapState=getRuntimeContext.getMapState(new MapStateDescriptor[String,Int]("MapStateDescriptor",createTypeInformation[String],createTypeInformation[Int]))
  }
  override def map(value: (String, String)): String = {
    var category:String = value._2
    //如果類別已經訪問過，訪問次數就在原有基礎上加1；如果沒有訪問過，就標記為1
    var count:Int=0
    if(mapState.contains(category)){
      count=mapState.get(category)
    }
    //把類別以及對應的訪問次數放入到狀態中
    mapState.put(category,count+1)
    //構建回傳值
    val list: List[String] = mapState.entries().asScala.map(entry => entry.getKey + ":" + entry.getValue).toList
    val str: String = list.mkString(" | ")
    value._1+"--->"+str
  }
}

3.4.4 需求：實作wordCount自動統計，練習ReducingState

import org.apache.flink.api.common.functions.{ReduceFunction, RichMapFunction, RuntimeContext}
import org.apache.flink.api.common.state.{ReducingState, ReducingStateDescriptor, ValueState, ValueStateDescriptor}
import org.apache.flink.api.common.typeinfo.TypeInformation
import org.apache.flink.api.java.tuple.Tuple
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala._
/**
 * 通過ReducingState實作wordcount自動統計
 */
object ReducingStateJob {
  def main(args: Array[String]): Unit = {
    /**
     * 1.執行環境
     * 2.資料源：socket
     * 3.資料處理：
     *    3.1 flatmap
     *    3.2 map--->(word,1)
     *    3.3 keyby   ===>dataStream轉換成了keyedStream
     *    3.4 map(new MyMapFunction)
     * 4.sink:print
     * 5.executeJob
     */
    /**
     * class MyMapFunction extends RichMapFunction
     * 通過valueState完成資料的統計處理
     * 1.在open方法中創建valueState物件
     *    a.需要RuntimeContext物件
     *
     *    b.RuntimeContext物件中提供的有方法，可以獲取到ValueState
     * 2.在map方法中使用valueState物件
     */
    val environment: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    val dataStream: DataStream[String] = environment.socketTextStream("192.168.229.10", 9999)
    val keyedStream: KeyedStream[(String, Int), Tuple] = dataStream.flatMap(_.split("\\s+"))
      .map((_, 1))
      .keyBy(0)
    val result: DataStream[String] = keyedStream.map(new MyReducingMapFunction)
    result.print()
    environment.execute("ReducingStateJob")
  }
}
/**
 * In：輸入資料的型別;根據使用這個函式的資料流（keyedStream）型別決定
 * Out：輸出資料的型別；map方法的回傳值型別，根據業務需要決定
 */
/*class MyMapFunction extends RichMapFunction[IN,Out]*/
class MyReducingMapFunction extends RichMapFunction[(String,Int),String]{
  //通過ReducingState完成wordcount 的自動統計
  var reducingState:ReducingState[Int]=_
  override def open(parameters: Configuration): Unit = {
    val context: RuntimeContext = getRuntimeContext
    val name:String="ReducingStateDescriptor"
    val typeInfo:TypeInformation[Int]=createTypeInformation[Int]
    val reduceFunction: ReduceFunction[Int] = new ReduceFunction[Int] {
      override def reduce(value1: Int, value2: Int): Int = {
//        print(value1+"****"+value2)
        value1+value2
      }
    }
    var reducingStateDescriptor:ReducingStateDescriptor[Int]=new ReducingStateDescriptor[Int](name,reduceFunction,typeInfo)
    reducingState=context.getReducingState(reducingStateDescriptor)
  }
  override def map(value: (String, Int)): String = {
    reducingState.add(value._2)//把需要計算的資料添加到reducingState里面
    value._1+":"+reducingState.get()
  }
}

3.4.5 需求：實作用戶訂單平均金額，練習AggeragetingState

import org.apache.flink.api.common.functions.{AggregateFunction, RichMapFunction}
import org.apache.flink.api.common.state.{AggregatingState, AggregatingStateDescriptor}
import org.apache.flink.api.common.typeinfo.TypeInformation
import org.apache.flink.api.java.tuple.Tuple
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala._
object AggregatingStateJob {
  def main(args: Array[String]): Unit = {
    val environment: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    //要求輸入的資料：  用戶編號  用戶名  訂單金額
    val dataStream: DataStream[String] = environment.socketTextStream("192.168.229.10", 9999)
    val keyedStream: KeyedStream[(String, Double), Tuple] = dataStream.map(_.split("\\s+"))
      .map(words => (words(0) + ":" + words(1), words(2).toDouble))
      .keyBy(0)
    val result: DataStream[String] = keyedStream.map(new MyAggregateMapFunction)
    result.print()
    environment.execute("AggregatingStateJob")
  }
}

//通過aggregatingState完成訂單的平均金額的計算
class MyAggregateMapFunction extends RichMapFunction[(String,Double),String]{
  //第一個Double表示的是訂單金額；第二個Double表示的是用戶的訂單平均金額
  var aggregatingState:AggregatingState[Double,Double]=_
  override def open(parameters: Configuration): Unit = {
    //第一個Double:輸入型別，就是訂單金額
    //第二個型別（Double,Int）:中間型別，計算程序中的型別，表示（訂單總金額,訂單個數）
    //第三個型別Double:輸出型別，就是訂單平均金額
    var name:String="aggregatingStateDescriptor"
    var aggFunction:AggregateFunction[Double,(Double,Int),Double]=new AggregateFunction[Double,(Double,Int),Double] {
      override def createAccumulator(): (Double, Int) = (0,0)//初始值
      /**
       * 中間計算程序
       * @param value 輸入資料，訂單金額
       * @param accumulator 中間計算結果 (訂單總金額,訂單個數)
       * @return
       */
      override def add(value: Double, accumulator: (Double, Int)): (Double, Int) = (accumulator._1+value,accumulator._2+1)
      //計算結果
      override def getResult(accumulator: (Double, Int)): Double = accumulator._1/accumulator._2
      override def merge(a: (Double, Int), b: (Double, Int)): (Double, Int) = (a._1+b._1,a._2+b._2)
    }
    var accType:TypeInformation[(Double,Int)]=createTypeInformation[(Double,Int)]
    var aggregatingStateDescriptor:AggregatingStateDescriptor[Double,(Double,Int),Double]= new AggregatingStateDescriptor[Double,(Double,Int),Double](name,aggFunction,accType)
    aggregatingState=getRuntimeContext.getAggregatingState[Double,(Double,Int),Double](aggregatingStateDescriptor)
  
  override def map(value: (String, Double)): String = {
    aggregatingState.add(value._2)//把這一次訂單的金額放進去
    val avg: Double = aggregatingState.get()//獲取到狀態中計算完成之后的訂單平均金額
    value._1+"的訂單平均金額："+avg
  }
}

轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/294153.html

標籤：其他

上一篇：dubbo學習筆記

下一篇：Elasticsearch斷路器相關知識及使用

Flink（四） 狀態管理 1