大資料——Flink 常見面試題整理-有解無憂

1. Flink 的特點

2. Flink 和 SparkStreaming 的對比

3. Flink 和 Blink、Alink之間的關系

4. JobManager 和 TaskManager 的職責

5. Flink 集群部署有哪些模式

6. Flink Dataflow 介紹

7. Parallelism 和 Slot 的理解

8. Flink 執行圖

9. DatastreamAPI 常見的Transform 算子

10. Flink 中視窗型別

11. Flink 中視窗函式的分類

12. Flink 中 window的功能

13. 談談對 Flink 中時間語意的理解

14. 談談對 Flink 中 Watermark 的理解

15. 談談對 Flink 狀態編程的理解

16. 談談 Flink 中是如何實作對狀態（state）的存盤、訪問及維護的（狀態后端——state backend）

17. Flink 檢查點演算法

18. Flink 的重啟策略

19. Flink 中 TableAPI 和 FlinkSQL 的基本使用，1.9 版本以后引入了 Blink Planner

20. 談談對于 Flink 中廣播狀態（broadcast State）的理解

21. Flink 中如何保證端到端的狀態一致性

22. 兩階段提交對 Sink 系統的要求

23. Flink 的監控主要看哪些指標（Mertics）

24. Flink 中如何實作反壓

25. Flink 的優化

26. Flink 中 CEP 的應用

1. Flink 的特點

Flink 具有高吞吐、低延時、高性能的特點
支持基于時間語意、視窗及狀態編程
同時還具備了 checkpoint 和 savepoint 的功能

2. Flink 和 SparkStreaming 的對比

Flink支持實時流處理，而 SparkStreaming 是通過微批處理的方式來實作實時處理，犧牲了吞吐量
Flink 支持狀態編程，而 SparkStreaming 不支持
Flink：JobManager、TaskManager SparkStreaming：master、worker

3. Flink 和 Blink、Alink之間的關系

Blink 是Flink的一個分支版本，由阿里團隊開發，在Flink 1.9版本之后，融入了Blink Table API 的Blink Planner
Alink 是基于 Flink 的通用演算法平臺，也是由阿里團隊開發，更多的是用于機器學習和人工智能，

4. JobManager 和 TaskManager 的職責

JobManager

協調 Flink 應用程式的分布式執行，它決定何時調度下一個 task（或一組 task）、對完成的 task 或執行失敗做出反應、協調 checkpoint、并且協調從失敗中恢復等等，這個行程由三個不同的組件組成：

ResourceManager：負責 Flink 集群中的資源提供、回收、分配，它管理 task
slots，這是 Flink 集群中資源調度的單位，Flink 為不同的環境和資源提供者（例如
YARN、Kubernetes 和 standalone 部署）實作了對應的 ResourceManager，
Dispatcher：提供了一個 REST 介面，用來提交 Flink 應用程式執行，并為每個提交的作業啟動一個新的 JobMaster，它還運行 Flink WebUI 用來提供作業執行資訊，
JobMaster：負責管理單個JobGraph的執行，Flink 集群中可以同時運行多個作業，每個作業都有自己的 JobMaster，

TaskManager

執行作業流的 task，并且快取和交換資料流，在 TaskManager 中資源調度的最小單位是 task slot，TaskManager 中 task slot 的數量表示并發處理 task 的數量，

對于分布式執行，Flink 將算子的 subtasks 鏈接成 tasks，每個 task 由一個執行緒執行，將算子鏈接成 task 是個有用的優化：它減少執行緒間切換、緩沖的開銷，并且減少延遲的同時增加整體吞吐量，

每個 task slot 代表 TaskManager 中資源的固定子集，例如，具有 3 個 slot 的 TaskManager，會將其托管記憶體 1/3 用于每個 slot，分配資源意味著 subtask 不會與其他作業的 subtask 競爭托管記憶體，而是具有一定數量的保留托管記憶體，注意此處沒有 CPU 隔離；當前 slot 僅分離 task 的托管記憶體，

通過調整 task slot 的數量，用戶可以定義 subtask 如何互相隔離，每個 TaskManager 有一個 slot，這意味著每個 task 組都在單獨的 JVM 中運行，具有多個 slot 意味著更多 subtask 共享同一 JVM，同一 JVM 中的 task 共享 TCP 連接（通過多路復用）和心跳資訊，它們還可以共享資料集和資料結構，從而減少了每個 task 的開銷，

5. Flink 集群部署有哪些模式

StandAlone、StandAlone HA、Flink on Yarn（Yarn Session、Pre-Job）

6. Flink Dataflow 介紹

Dataflow：Flink 程式在執行的時候會被映射成一個資料流模型

Operator：資料模型中每一個操作被稱為 Operator，Operator 可以被分為 Source、Transform、Sink

Operator Chain：客戶端在提交任務的時候會對 Operator 進行優化，如果 Operator 在資料傳遞程序中保持資料的磁區數和資料的排序不變，則會將 Operator 進行合并，合并后被稱為 Operator Chain，實際上就是一個執行鏈，每個執行鏈會在 TaskManager 上的一個獨立的執行緒中執行，即 SubTask

Partition：資料流模型是分布式的和并行的，執行中會形成1~n個磁區

SubTask：多個磁區任務可以并行，每一個都是獨立運行在一個執行緒中，也就是一個 Subtask 子任務

Parallelism：并行度，就是可一個同時真正執行的子任務數/磁區數

7. Parallelism 和 Slot 的理解

Slot 是靜態的概念，是指 TaskManager 具有的并發執行能力

Parallelism 是動態的概念，是指程式運行時實際使用的并發能力

設定合適的 Parallelism 能提高運算效率，太多和太少都不行，正常情況下 Parallelism <= Slot

設定Parallelism有多種方式，優先級為 API > env > p > file

8. Flink 執行圖

Flink 中執行圖可以分為4層：StreamGraph -> JobGraph -> ExecutionGraph -> 物理執行圖

StreamGraph：根據用戶通過 Stream API 撰寫的代碼生成的最初的圖，表示程式的拓撲結構

JobGraph：StreamGraph 經過優化后生成了 JobGraph 提交給 JobManager 的資料結構，主要優化是將多個符合條件的節點 chain 在一起作為一個節點（Operator Chain），這樣可以減少資料在節點之間流動所需要的序列化/反序列化/傳輸消耗

ExecutionGraph：JobManager 根據 JobGraph 生成 ExecutionGraph，是 JobGraph的并行化版本，是調度層最核心的資料結構

物理執行圖：JobManager 根據 ExecutionGraph 對 Job 進行調度后，在各個 TaskManager 上部署 Task 后形成的“圖”，并不是一個具體的資料結構

9. Flink 代碼實作步驟（以 wordcount 為例）

//1. 創建環境（Execution Environment）
val env = SteamExecutionEnvironment.getExecutionEnvironment

//2. 加載資料源（Source）
val dataStream = env.readTextFile("D:\\study\\wordcount\\input")

//3. 資料轉換（Transform）
val result = dataStream.flatMap(_.split(" ")).map((_,1)).keyBy(0).sum(1)

//4. 結果輸出（Sink）
result.print()

//5. 執行程式(Execute)
env.execute()

9. DatastreamAPI 常見的Transform 算子

單流：

Map
FlatMap
Filter
KeyBy
Reduce
Aggregations
max/maxBy min/minBy

多流：

union
connect
join
coMap/coFlatMap
process

10. Flink 中視窗型別

針對上游資料是keyedStream還是非keyedStream可以分為Keyed Windows 和 Non-Keyed Windows

基于數量的 CountWindow

//滾動視窗
//指的是同一個 key 的資料達到5個才執行，不是總共的資料達到5
streamKeyBy.countWindow(5)

//滑動視窗
//指定步長為2，當單個 key 滿足2個時就執行一次，但是計算的視窗是大小為5的值，而這里的視窗大小指的也是同一個 key 達到5條
streamKeyBy.countWindow(5,2)

基于時間的 TimeWindow

//滾動視窗
//每5秒一個視窗
streamKeyBy.window(TumblingProcessingTimeWindows.of(Time.seconds(5)))

//滑動視窗
//每10秒一個視窗，5秒的滑動，5秒執行一次，計算的是10秒的資料
streamKeyBy.window(SlidingProcessingTimeWindows.of(Time.seconds(10),Time.seconds(5)))

//會話視窗
//會話視窗在時間視窗中使用，如果視窗中沒資料的話就不觸發執行
streamKeyBy.window(ProcessingTimeSessionWindows.withGap(Time.seconds(5)))

11. Flink 中視窗函式的分類

增量聚合函式：

每條資料到來就進行計算，保持一個簡單的狀態，場景：求和

ReduceFunction（reduce）、AggregateFunction（aggregate）

全視窗函式：

先把視窗所有資料收集起來，等計算的時候遍歷所有資料，場景：視窗內排序

相比于增量聚合函式，全視窗函式能夠取到更多的背景關系資訊，例如視窗資訊，狀態的資訊

ProcessWindowFunction（process）、WindowFunction（apply）

案例

12. Flink 中 window的功能

reduce、aggregate：對視窗內的資料進行聚合計算
process、apply：可以獲得視窗內的背景關系資訊，如視窗資訊和狀態資訊
assigner：分配器，將資料流中的元素分配到對應的視窗
trigger：觸發器，定義 window 什么時候觸發計算
evictor：剔除器，定義移除某些資料的邏輯
allowedLateness：允許處理遲到的資料
sideOutputLateData：將遲到的資料放入側輸出流
getSideOutput：獲取側輸出流

13. 談談對 Flink 中時間語意的理解

三種時間概念：事件時間、接入時間、處理時間

事件時間：EventTime，資料產生的時間
接入時間：IngestionTime，資料進入 Flink 的時間
處理時間：ProcessTime，資料被算子處理的時間

關于視窗起始時間的計算值

左閉右開

timestamp-(timestamp-offset+windowSize)%windowSize

14. 談談對 Flink 中 Watermark 的理解

為什么引入 Watermark

由于網路或者系統等外部因素的影響，資料被傳輸到 Flink 的時間往往不是按照事件產生的順序傳輸過來的，因而會造成亂序或者延遲等問題，在此情況下，引入了 Watermark 機制，用于衡量資料到達的進度和完整性

Watermark 的計算

Flink 將最新讀取資料的最大的 EventTime 減去固定的時間間隔作為 Watermark，固定的時間間隔其實就是指最大延遲時間，如果有一條資料的 Watermark 大于了某個視窗的 EndTime，就會默認該視窗內的資料已經全部到達并觸發執行

package org.example.window
import java.time.Duration

import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, WatermarkStrategy}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.assigners.{EventTimeSessionWindows, SlidingEventTimeWindows, TumblingEventTimeWindows}
import org.apache.flink.streaming.api.windowing.time.Time
import org.example.bean.TrainAlarm

 //設定eventTime 和 watertime 處理亂序時間
 
object AssignEventTimeAndWm {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    val inputStream= env.socketTextStream("master", 666)
      .map(line => {
        val ps = line.split(",")
        TrainAlarm(ps(0), ps(1).toLong, ps(2).toDouble)
      })
      .assignTimestampsAndWatermarks(
      //Duration 設定延遲時長 watermark = 當前已經到達的最大eventTime - 延時時長
      //只要比watermark小的視窗就可以觸發
      WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(5))
        .withTimestampAssigner(new SerializableTimestampAssigner[TrainAlarm] {
          //設定eventTime是哪個欄位
          override def extractTimestamp(element: TrainAlarm, l: Long): Long = {
            element.ts*1000L
          }
        })
    )

    inputStream.keyBy(_.id)
      .window(TumblingEventTimeWindows.of(Time.seconds(5)))
      .max("temp")
      .print()

     env.execute()
  }
}

案例

15. 談談對 Flink 狀態編程的理解

狀態：可以理解為是資料流在計算處理的中間結果，一般接個富函式或者 ProcessFunction 獲取狀態

Flink 會進行狀態管理，包括狀態一致性、故障處理以及高效存盤和訪問，以便開發人員可以專注于應用程式的邏輯
在 Flink 中，狀態由每一個 Task 維護，狀態始終與特定的算子相關聯
狀態的型別
- 算子狀態：算子狀態的作用范圍限定為算子任務
  - 串列狀態（List state）
  - 聯合串列狀態（Union list state）
  - 廣播狀態（Broadcast state）
- 鍵控狀態：根據輸入資料流中定義的鍵（key）來維護和訪問，不同的 key 維護自己的狀態，并且不同 key 的狀態不同
  - 值狀態（Value state）：將狀態表示為單個值
  - 串列狀態（List state）：將狀態表示為一組資料的串列
  - 映射狀態（Map state）：將狀態表示為一組Key-Value對
  - 聚合狀態（Reducing state&Aggregate state）：將狀態表示為一個用于聚合的操作，將一個新到的值直接帶入進去做聚合操作


package org.example.state

import org.apache.flink.api.common.state.ValueStateDescriptor
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector
import org.example.bean.TrainAlarm

/**
 * 10 s 內溫度連續上升就報警
 * 定義三個狀態：溫度狀態、時間狀態、個數狀態
 * 如果是第一條資料，更新溫度狀態值、注冊10s后觸發的定時器并更行時間狀態，個數狀態設定為1
 * 如果不是第一條資料，
 *    如果溫度比溫度狀態的值大，更新溫度狀態值，個數狀態設定 +1
 *    如果溫度比溫度狀態的值小，洗掉定時器（從時間狀態中取時間）個數狀態設定為1，重新注冊定時器，更新溫度狀態值
 */
object TrainTempAlarmWithState2 {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val rst = env.socketTextStream("master", 666)
      .map(line => {
        val ps = line.split(",")
        TrainAlarm(ps(0), ps(1).toLong, ps(2).toDouble)
      }).assignAscendingTimestamps(_.ts*1000L)
      .keyBy(_.id)
      .process(new TempRiseWithTime())
      .print()
    env.execute()
  }
}
class TempRiseWithTime extends KeyedProcessFunction[String,TrainAlarm,String]{
  //定義三個狀態：溫度狀態、時間狀態、個數狀態
  lazy val tempState = getRuntimeContext.getState[Double](new ValueStateDescriptor[Double]("tempstate",classOf[Double]))
  lazy val timeState = getRuntimeContext.getState[Long](new ValueStateDescriptor[Long]("timestate",classOf[Long]))
  lazy val countState = getRuntimeContext.getState[Int](new ValueStateDescriptor[Int]("countstate",classOf[Int]))
  override def processElement(value: TrainAlarm, ctx: KeyedProcessFunction[String, TrainAlarm, String]#Context, out: Collector[String]): Unit = {
    if (tempState.value()==0 || timeState.value() ==0){
      tempState.update(value.temp)
      ctx.timerService().registerEventTimeTimer(value.ts*1000L+10000L)
      timeState.update(value.ts*1000L+10000L)
      countState.update(1)
    }else{
      if (value.temp>= tempState.value()){
        tempState.update(value.temp)
        countState.update(countState.value()+1)
      }else{
        ctx.timerService().deleteEventTimeTimer(timeState.value())
        countState.update(1)
        ctx.timerService().registerEventTimeTimer(value.ts*1000L+10000L)
        timeState.update(value.ts*1000L+10000L)
        tempState.update(value.temp)
      }
    }
  }

  override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[String, TrainAlarm, String]#OnTimerContext, out: Collector[String]): Unit = {
    if(countState.value()>=2){
      out.collect(ctx.getCurrentKey + " is alarming")
      tempState.clear()
      timeState.clear()
      countState.clear()
    }
  }
}

16. 談談 Flink 中是如何實作對狀態（state）的存盤、訪問及維護的（狀態后端——state backend）

狀態后端主要負責的兩件事：

本地狀態的管理
將檢查點（checkpoint）狀態寫入遠程存盤

Flink 提供的三種狀態后端：

MemoryStateBackend：存盤在TaskManager記憶體中，速度快，已丟失------生產環境基本不用
FsStateBackend：將checkpoint存盤到持久化檔案系統（FileSystem），而對于本地狀態，也會存到TaskManager記憶體中，還是會受記憶體溢位（OOM）影響
RockDBstateBackend：將所有狀態序列化后，存入到本地的RockDB中，而RockDB是基于KV的，可以看作是一個本地資料庫（實際使用記憶體+磁盤），checkpoint存到持久化檔案系統（FileSystem）上，

另外，如果設定的是 RockDBStateBackend，需要先引入依賴

<dependency> 
    <groupId>org.apache.flink</groupId> 
    <artifactId>flink-statebackend-rocksdb_2.12</artifactId> 
    <version>${flink.version}</version>
 </dependency>

17. Flink 檢查點演算法

一般檢查點實作方式：暫停應用，保存檢查點，再重新恢復應用

Flink 實作了基于 Chandy-Lamport 演算法的分布式快照，將檢查點的保存和資料處理分離開，不暫停整個應用

具體實作方式：

Flink 會在輸入資料集上間隔性的生成 Checkpoint barrier，通過柵欄（barrier）將間隔時間段內的資料劃分到對應的 checkpoint 中

對于 barrier 已經到達的磁區，繼續到達的資料會被緩沖，暫時不會被處理

18. Flink 的重啟策略

作用：可以控制在發生故障是如何重新啟動作業

如果未啟用檢查點，則使用“無重啟”策略，如果激活了檢查點并且尚未配置重啟策略，會采用“固定延遲”策略Integer.MAX_VALUE嘗試重啟

重啟策略分為4種：固定延遲重啟策略、故障率重啟策略、無重啟策略、后備重啟策略

固定延遲重啟策略：

嘗試給定重啟作業的次數，如果超過最大嘗試次數則作業失敗，兩次連續重啟嘗試之間，會有一個固定的延遲等待時間

通過在 flink-conf.yaml 中配置引數：

# fixed-delay:固定延遲策略
restart-strategy: fixed-delay

# 嘗試5次，默認Integer.MAX_VALUE
restart-strategy.fixed-delay.attempts: 5

# 設定延遲時間10s，默認為 akka.ask.timeout時間
restart-strategy.fixed-delay.delay: 10s

通過代碼中修改：

val env = StreamExecutionEnvironment.getExecutionEnvironment

//設定最大嘗試次數為5，重啟間隔為10秒
env.setRestartStrategy(RestartStrategies.fixedDelayRestart(5,Time.seconds,10))

故障率重啟策略：

在故障后重新作業，當設定的故障率（failure rate）超過每個時間間隔的故障是，作業最終失敗，在兩次連續嘗試之間，策略延遲等待一段時間

在 flink-conf.yaml 中配置引數：

# 設定重啟策略為failure-rate
restart-strategy: failure-rate

# 失敗作業之前的給定時間間隔內的最大重啟次數，默認1
restart-strategy.failure-rate.max-failures-per-interval: 3

# 測量故障率的時間間隔，默認1min
restart-strategy.failure-rate.failure-rate-interval: 5min

# 兩次連續重啟嘗試之間的延遲，默認akka.ask.timeout時間
restart-strategy.failure-rate.delay: 10s

在代碼中設定：

val env = StreamExecutionEnvironment.getExecutionEnvironment();

// 3為最大失敗次數；5min為測量的故障時間；10s為2次間的延遲時間
env.setRestartStrategy(RestartStrategies.failureRateRestart(3,Time.minutes(5),Time.seconds(10)));

無重啟策略：

作業直接失敗，不嘗試重啟

在 flink-conf.yaml 中配置：

restart-strategy: none

在代碼中設定：

val env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setRestartStrategy(RestartStrategies.noRestart());

后備重啟策略：

使用集群定義的重新啟動策略，這對于啟用檢查點的流式傳輸程式很有幫助，

19. Flink 中 TableAPI 和 FlinkSQL 的基本使用，1.9 版本以后引入了 Blink Planner

使用 TableAPI 和 FlinkSQL需要匯入的依賴

<dependency> 
	<groupId>org.apache.flink</groupId> 
	<artifactId>flink-table-api-scala-bridge_2.12</artifactId> 
	<version>${flink.version}</version> 
</dependency> 
<dependency> 
	<groupId>org.apache.flink</groupId> 
	<artifactId>flink-table-planner-blink_2.12</artifactId> 
	<version>${flink.version}</version> 
</dependency> 
<dependency> 
	<groupId>org.apache.flink</groupId> 
	<artifactId>flink-table-common</artifactId> 
	<version>${flink.version}</version> 
</dependency> 
<dependency> 
	<groupId>org.apache.flink</groupId> 
	<artifactId>flink-csv</artifactId> 
	<version>${flink.version}</version> 
</dependency>
	 <dependency> 
	 <groupId>org.apache.flink</groupId>
	 <artifactId>flink-json</artifactId> 
	 <version>${flink.version}</version> 
 </dependency>

程式結構

創建表環境

	val bsEnv = StreamExecutionEnvironment.getExecutionEnvironment
    bsEnv.setParallelism(1)
    //推薦使用BlinkPlanner
    val bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build()
    val bsTableEnv = StreamTableEnvironment.create(bsEnv, bsSettings)

連接外部資料源，創建輸入表

tableEnv.executeSql("CREATE TEMPORARY TABLE table1 ... WITH ( 'connector' = ... )")

連接外部資料輸出，創建輸出表

tableEnv.executeSql("CREATE TEMPORARY TABLE outputTable ... WITH ( 'connector' = ... )")

TableAPI實作查詢操作

 val table2 = tableEnv.from("table1").select(...)

FlinkSQL實作查詢操作

val table3 = tableEnv.sqlQuery("SELECT ... FROM table1 ...")

將查詢結果放入輸出表中

val tableResult = table2.executeInsert("outputTable") tableResult...

TableAPI 和 FlinkSQL 的使用，官網給出的案例非常詳細，在寫代碼時可以借鑒：https://ci.apache.org/projects/flink/flink-docs-release-1.13/zh/

20. 談談對于 Flink 中廣播狀態（broadcast State）的理解

簡單理解：一個地吞吐量流包含了一組規則，我們向對來自另一個留的所有元素基于及規則進行評估計算，

場景：動態更新計算規則、大小表關聯

與其他運算子狀態的區別：

（1）它有一個map格式，用于定義存盤結構

（2）它僅對具有廣播流和非廣播流輸入的特定運算子可用

（3）這樣的運算子可以具有不同名稱的多個廣播狀態

操作流程

21. Flink 中如何保證端到端的狀態一致性

什么是狀態一致性：

有狀態的流處理，每個算子任務都可以有自己的狀態，所謂的狀態一致性，其實就是我們所說的計算結果要保證準確，一條資料不應該被丟失，也不應該被重復計算，在遇到故障時可以恢復狀態，恢復以后得重新計算，結果應該也是完全正確的，

狀態一致性分類：

At-Most-Once（最多一次）：

當任務故障時，最簡單的做法就是什么都不干，既不恢復丟失的資料，也不重復資料，最多處理一次事件，資料可能會丟失，但是處理的速度快，

At-Least-Once（至少一次）：

在大多數的真實應用場景，我們不希望資料丟失，所有的事件都會被處理，而且可以被多次處理，

Exactly-Once（精確一次）：★★★★★

恰好保證每個事件只被處理了一次，既沒有資料丟失，也沒有資料重復處理的情況出現

端到端的 Exactly-Once：★★★★★

內部保證：checkpoint
Source 端：可重置資料的讀取位置，比如 kafka 的偏移量可以手動維護，提交，
Sink 端：從故障恢復時，資料不會重復寫入外部系統，（冪等寫入、事務寫入）

注：
冪等寫入：就是說一個操作，可以重復執行很多次，但只導致一次結果更改，后面再重復執行就不起作用了，
事務寫入：原子性，一個事務中的一系列操作，要么全部成功，要么一個不做，
實作的思想，構建的事務對應著 checkpoint，等到 checkpoint 真正完成的時候，才把所有對應的結果寫入 Sink 系統中，實作方式，預寫日志（GenericWriteAheadSink）和兩階段提交（TwoPhaseCommitSinkFunction），
案例：
Flink 與 Kafka 端到端的 Exactly-Once：

Flink 內部：利用 checkpoint 機制，把狀態存盤，發生故障時可以恢復，保證內部的狀態一致性，
Source：KafkaConsumer 作為 Source，可以將偏移量作為狀態保存下來，如果后續任務發現了故障，恢復的時候可以由連接器重置偏移量，重新消費資料，保證一致性，
Sink ： KafkaProducer 作為 Sink ，采用兩階段提交 Sink ，需要實現TwoPhaseCommitSinkFunction，

package cn.jixiang.checkpoint

import java.lang
import java.util.{Properties, Random}

import org.apache.flink.api.common.RuntimeExecutionMode
import org.apache.flink.api.common.restartstrategy.RestartStrategies
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.api.common.time.Time
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend
import org.apache.flink.streaming.api._
import org.apache.flink.streaming.api.environment.CheckpointConfig.ExternalizedCheckpointCleanup
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer.Semantic
import org.apache.flink.streaming.connectors.kafka.{FlinkKafkaConsumer, FlinkKafkaProducer, KafkaSerializationSchema}
import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.kafka.clients.producer.ProducerRecord

/***
  * 從Kafka讀取資料，實作WC，寫回到Kafka，
  * 實作端到端的狀態一致性保證，
  */
object End2EndExactlyOnce {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC)

    // 設定checkpoint
    val hashMapStateBackend = new HashMapStateBackend()
    env.setStateBackend(new HashMapStateBackend())
    env.getCheckpointConfig.setCheckpointStorage("file:///D:\\Note\\Projects\\02\\Flink\\cha01\\ckp")
    env.enableCheckpointing(1000,CheckpointingMode.EXACTLY_ONCE)
    env.getCheckpointConfig.setMinPauseBetweenCheckpoints(500)
    env.getCheckpointConfig.setCheckpointTimeout(60000)
    env.getCheckpointConfig.setTolerableCheckpointFailureNumber(10)
    env.getCheckpointConfig.enableExternalizedCheckpoints(ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION)
    env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,Time.milliseconds(600)))

    // 從Kafka讀取資料
    val props1 = new Properties()
    props1.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "master:9092")
    props1.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "group-1")
    props1.setProperty(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "2000")
    props1.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true")
    props1.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest")

    val kafkaSource = new FlinkKafkaConsumer[String]("test1",new SimpleStringSchema(),props1)
    //提交offset到kafka
    kafkaSource.setCommitOffsetsOnCheckpoints(true)
    val inputData: DataStream[String] = env.addSource(kafkaSource)

    // transformation轉換
    val result = inputData
      .flatMap(_.split(" "))
      .map(t => {
        val random = new Random()
        val num = random.nextInt(5)
        if (num == 2){
          println(num)
          throw new Exception("哎呀呀，是例外呀")
        }
        (t,1)
      })
      .keyBy(_._1)
      .sum(1)
      .map(t => t._1 + ":" + t._2)

    // 往Kafka寫入資料
    val props2 = new Properties()
    props2.setProperty("bootstrap.servers", "master:9092")
    // 默認情況下，Kafka broker 將 transaction.max.timeout.ms 設定為 15 分鐘，
    // 此屬性不允許為大于其值的 producer 設定事務超時時間，
    // 默認情況下，FlinkKafkaProducer 將 producer config 中的 transaction.timeout.ms 屬性設定為 1 小時
    // 因此在使用 Semantic.EXACTLY_ONCE 模式之前應該增加 transaction.max.timeout.ms 的值，
    props2.setProperty("transaction.timeout.ms",1000*60*5+"")

    val myProducer = new FlinkKafkaProducer[String]("test2",
      new KafkaSerializationSchema[String]() {
        override def serialize(element: String, timestamp: lang.Long): ProducerRecord[Array[Byte], Array[Byte]] = {
          new ProducerRecord[Array[Byte], Array[Byte]]("test2",element.getBytes,element.getBytes("utf-8"))
        }
      },
      props2,
      Semantic.EXACTLY_ONCE
    )

    result.print()
    result.addSink(myProducer)

    env.execute("Flink + Kafka")
  }
}

22. 兩階段提交對 Sink 系統的要求

外部系統必須提供事務支持，Kafka、Mysql

原因：

在 checkpoint 隔離期間，必須開啟事務并接收資料寫入，

在收到 checkpoint 完成的通知之前，事務必須是“等待提交狀態”，如果在此狀態下 sink 系統關閉了事務（例如超時），則未提交的資料就會丟失

Sink 任務必須能夠在行程失敗后恢復事務（利用了事務的持久性中的 rollback 機制）

提交事務必須是冪等操作（事務的一致性）

23. Flink 的監控主要看哪些指標（Mertics）

Counter：對流處理的資料進行累加計數
Guage：可以反映一個值，比如查看記憶體使用情況
Meter：值統計吞吐量和時間單位內發生“事件”的次數，計算方式：事件次數除以使用的時間
Histogram：用于統計一些資料的分布，如Quantile、Mean、StdDev、Max、Min等

24. Flink 中如何實作反壓

什么是反壓：

訊息處理速度 < 訊息的發送速度，訊息擁堵，系統運行不暢，通過Consumer 給 Producer 一個反饋，告知所能接受資料的大小，從而使 Producer 減少發送資料的頻率

反壓的影響：

checkpoint 的時長：checkpoint barrier 跟隨普通資料流動，如果資料處理被阻塞，使得 checkpoint barrier 流經整個資料管道的時長變長，導致 checkpoint 總體時間變長
state 大小：為保證 Exactly-Once 準確一次，對于有兩個以上輸入管道的 Operator，checkpoint barrier 需要對齊，即接受到較快的輸入管道的 barrier 后，它后面資料會被快取起來但不處理，直到較慢的輸入管道的 barrier 也到達，這些被快取的資料會被放到 state 里面，導致 checkpoint 變大，

反壓機制：

Flink1.5 之前是基于 TCP 的反壓機制

弊端：
- 單個 Task 導致的反壓，會阻斷整個 TM-TM 的 socket，連checkpoint barrier 也無法發出
- 反壓傳播路徑太長，導致生效延遲較大

Flink1.5 之后采用 Credit-base 反壓機制

資料寫入端將資料寫入到buffer中，
判斷當前的credit值是否大于0，
如果credit > 0，則將資料寫出，并更新credit值，資料寫出的時候會在msg上帶上當前生成端的資料量backlog，
如果credit <= 0，則不寫

25. Flink 的優化

記憶體管理、資料去重（使用布隆過濾器）、資料傾斜、checkpoint優化、代碼重用

參考《大資料—— Flink 的優化》

26. Flink 中 CEP 的應用

什么是CEP：復雜事件處理（Complex Event Processing）

實作方式：將資料流通過一定的規則匹配（模式），然后輸出用戶想得到的資料

使用場景：

風控檢測：對用戶例外行為模式、資料例外流向實時檢測
策略營銷：向特定行為的用戶進行濕濕的精準營銷
運維監控：監控設備運行引數，靈活配置多指標的發生規則

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

標籤：其他

上一篇：①?全網首發??資料有道之資料庫技術??干貨大全【持續更新】???

下一篇：看完還不會資料庫優化，你來找我！