1. 簡單介紹一下Flink

Flink是一個面向流處理和批處理的分布式數據計算引擎，能夠基于同一個Flink 運行，可以提供流處理和批處理兩種類型的功能。

在 Flink 的世界觀中，一切都是由流組成的，離線數據是有界的流；實時數據是一個沒有界限的流，這就是所謂的有界流和無界流。

Flink的核心特性:支持有狀態流處理、事件時間語義、精確一次容錯（Checkpoint機制）、批流一體架構。

 

2. Flink的運行必須依賴Hadoop組件嗎？

Flink可以完全獨立于Hadoop，在不依賴Hadoop組件下運行。但是做為大數據的基礎設施，Hadoop體系是任何大數據框架都繞不過去的。Flink可以集成眾多 Hadooop 組件，例如Yarn、Hbase、HDFS等等。例如，Flink可以和Yarn集成做資源調度，也可以讀寫HDFS，或者利用HDFS做檢查點。

 

3. Flink集群運行時角色

Flink 運行時由兩種類型的進程組成：一個 JobManager 和一個或者多個TaskManager。

Client 不是運行時和程序執行的一部分，而是用于準備數據流并將其發送給JobManager。之后，客戶端可以斷開連接（分離模式），或保持連接來接收進程報告（附加模式）。客戶端可以作為觸發執行 Java/Scala 程序的一部分運行，也可以在命令行進程 ./bin/flink run … 中運行。

可以通過多種方式啟動 JobManager 和 TaskManager：直接在機器上作為standalone 集群啟動、在容器中啟動、或者通過YARN等資源框架管理并啟動。

TaskManager 連接到 JobManagers，宣布自己可用，并被分配工作。

(1). JobManager：

JobManager 具有許多與協調 Flink 應用程序的分布式執行有關的職責：它決定何時調度下一個 task（或一組 task）、對完成的 task 或執行失敗做出反應、協調 checkpoint、并且協調從失敗中恢復等等。這個進程由三個不同的組件組成：

①.ResourceManager

ResourceManager 負責 Flink 集群中的資源提供、回收、分配，管理 taskslots。

②.Dispatcher

Dispatcher 提供了一個 REST 接口，用來提交 Flink 應用程序執行，并為每個提交的作業啟動一個新的 JobMaster。它還運行 Flink WebUI 用來提供作業執行信息。

③.JobMaster

JobMaster 負責管理單個JobGraph的執行。Flink 集群中可以同時運行多個作業，每個作業都有自己的 JobMaster。

(2). TaskManagers：

TaskManager（也稱為 worker）執行作業流的 task，并且緩存和交換數據流。必須始終至少有一個 TaskManager。在 TaskManager 中資源調度的最小單位是 taskslot。TaskManager 中 taskslot 的數量表示并發處理 task 的數量。請注意一個 task slot 中可以執行多個算子。

 

4. Flink相比Spark Streaming有什么區別？

這個問題是一個非常宏觀的問題，因為兩個框架的不同點非常之多。但是在面試時有非常重要的一點一定要回答出來：Flink 是標準的實時處理引擎，基于事件驅動。而 Spark Streaming 是微批（Micro-Batch）的模型。

(1).架構模型

SparkStreaming 在運行時的主要角色包括：Master、Worker、Driver、Executor。

Flink 在運行時主要包含：Jobmanager、Taskmanager 和 Slot。

(2).任務調度

SparkStreaming 連續不斷的生成微小的數據批次，構建有向無環圖 DAG，Spark Streaming 會依次創建 DStreamGraph、JobGenerator、JobScheduler。

Flink 根據用戶提交的代碼生成 StreamGraph，經過優化生成 JobGraph，然后提交給 JobManager 進行處理，JobManager 會根據 JobGraph 生成 ExecutionGraph，ExecutionGraph 是 Flink 調度最核心的數據結構， JobManager 根據 ExecutionGraph 對 Job 進行調度。

(3).時間機制

SparkStreaming 支持的時間機制有限，只支持處理時間。

Flink 支持了流處理程序在時間上的三個定義：處理時間、事件時間、注入時間。同時也支持 watermark 機制來處理滯后數據。

(4).容錯機制

對于 SparkStreaming 任務，我們可以設置 checkpoint，然后假如發生故障并重啟，我們可以從上次 checkpoint 之處恢復，但是這個行為只能使得數據不丟失，可能會重復處理，不能做到恰一次處理語義。

Flink 則使用checkpoint機制（分布式快照）和兩階段提交協議來解決這個問題。

 

5. 介紹下Flink的容錯機制（checkpoint）

Checkpoint機制是Flink可靠性的基石，可以保證Flink集群在某個算子因為某些原因(如 異常退出)出現故障時，能夠將整個應用流圖的狀態恢復到故障之前的某一狀態，保證應用流圖狀態的一致性。Flink的Checkpoint機制原理來自“Chandy-Lamport algorithm”算法。

每個需要Checkpoint的應用在啟動時，Flink的JobManager為其創建一個CheckpointCoordinator(檢查點協調器)，CheckpointCoordinator全權負責本應用的快照制作。

CheckpointCoordinator(檢查點協調器) 周期性的向該流應用的所有 source算子發送 barrier(屏障)。

當某個source算子收到一個barrier時，便暫停數據處理過程，然后將自己的當前狀態制作成快照，并保存到指定的持久化存儲中，最后向CheckpointCoordinator報告自己快照制作情況，同時向自身所有下游算子廣播該barrier，恢復數據處理下游算子收到barrier之后，會暫停自己的數據處理過程，然后將自身的相關狀態制作成快照，并保存到指定的持久化存儲中，最后向CheckpointCoordinator報告自身快照情況，同時向自身所有下游算子廣播該barrier，恢復數據處理。

每個算子按照步驟3不斷制作快照并向下游廣播，直到最后barrier傳遞到sink算子，快照制作完成。

當CheckpointCoordinator收到所有算子的報告之后，認為該周期的快照制作成功; 否則，如果在規定的時間內沒有收到所有算子的報告，則認為本周期快照制作失敗。

 

6. Flink checkpoint與Spark Streaming的有什么區別或優勢嗎？

Spark streaming 的 checkpoint 僅僅是針對 driver 的故障恢復做了數據和元數據的 checkpoint。

而 Flink 的 checkpoint 機制 要復雜了很多，它采用的是輕量級的分布式快照，實現了每個算子的快照，及流動中的數據的快照。

 

7. Flink是如何保證Exactly-once語義的

Flink 實現 Exactly-once 語義的核心機制基于 分布式快照（Checkpoint） 和 兩階段提交協議（2PC），結合對 Source 和 Sink 的端到端一致性要求。
以下是詳細實現原理和關鍵步驟：

一、核心機制：分布式快照（Checkpoint）
1. Checkpoint 流程
觸發機制：JobManager 定期（可配置）向所有算子發送觸發 Checkpoint 的信號。
Barrier 插入：Source 接收到信號后，在數據流中插入 Checkpoint Barrier（特殊標記），將數據流邏輯劃分為多個 Checkpoint 周期。

狀態對齊：
算子收到 Barrier 后暫停處理后續數據，將當前狀態 異步持久化 到外部存儲（如 HDFS、RocksDB）。
確保 Barrier 之前的所有數據已處理完畢，狀態一致。
確認機制：所有算子完成狀態持久化后，向 JobManager 發送確認，標記該 Checkpoint 完成。

2. 容錯恢復
故障檢測：JobManager 監控 TaskManager 心跳，發現故障后觸發恢復。
狀態回滾：從最近一次成功的 Checkpoint 恢復所有算子的狀態，并重置 Source 的消費偏移量（如 Kafka Offset），實現 狀態一致性。

二、端到端 Exactly-once 的附加條件
僅靠 Checkpoint 無法保證端到端一致性，需 Source 和 Sink 配合：
1. Source 要求：可重放（Replayable）
示例：Apache Kafka 作為 Source，支持按 Offset 重新消費數據。
機制：Checkpoint 中保存 Kafka Offset，恢復時重新消費該 Offset 后的數據。

2. Sink 要求：事務性寫入
兩階段提交（2PC）：
預提交階段：Checkpoint 開始時，Sink 將數據寫入外部系統的 事務緩沖區（如 Kafka 事務消息）。
提交階段：Checkpoint 完成后，提交事務（如 Kafka 事務提交），確保數據對外可見。
回滾機制：若 Checkpoint 失敗，事務回滾，數據不對外暴露。

3. 典型支持組件
Source：Kafka、Kinesis（支持 Offset 重置）。
Sink：Kafka（事務生產者）、支持 2PC 的數據庫（如 TiDB）、HDFS（通過重命名文件實現原子性）。

三、關鍵配置與優化
1. Checkpoint 參數
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 啟用 Checkpoint（間隔 5 分鐘）
env.enableCheckpointing(5 * 60 * 1000);
// 設置 Checkpoint 模式為 Exactly-once
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
// 超時時間（超過則丟棄）
env.getCheckpointConfig().setCheckpointTimeout(10 * 60 * 1000);
// 最大并發 Checkpoint 數
env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);

2. 狀態后端選擇
RocksDBStateBackend：適合大狀態場景，支持增量 Checkpoint，減少每次持久化數據量。
FsStateBackend：狀態存于文件系統（如 HDFS），適合中小規模狀態。

3. 端到端配置示例（Kafka + Flink + Kafka）
// Source 配置（Kafka 可重放）
FlinkKafkaConsumer<String> source = new FlinkKafkaConsumer<>(
"input-topic",
new SimpleStringSchema(),
properties
);
source.setStartFromLatest();

// Sink 配置（Kafka 事務生產者）
FlinkKafkaProducer<String> sink = new FlinkKafkaProducer<>(
"output-topic",
new SimpleStringSchema(),
properties,
FlinkKafkaProducer.Semantic.EXACTLY_ONCE // 啟用 Exactly-once 模式
);

四、運維視角的注意事項
1、監控 Checkpoint 成功率：
通過 Flink Web UI 或 Metrics Reporter 監控 numberOfCompletedCheckpoints 和 numberOfFailedCheckpoints。
失敗常見原因：反壓（處理速度跟不上 Checkpoint 間隔）、網絡抖動、存儲系統不可用。

2、避免 Checkpoint 超時：
增大 checkpointTimeout 或優化作業性能（如減少算子鏈長度、調整并行度）。

3.資源預留：
Checkpoint 期間需要額外網絡和磁盤 IO，確保 TaskManager 資源充足。

 

8. 如果下級存儲不支持事務，Flink怎么保證exactly-once

端到端的exactly-once對sink要求比較高，具體實現主要有冪等寫入和事務性寫入兩種方式。

冪等寫入的場景依賴于業務邏輯，更常見的是用事務性寫入。而事務性寫入又有預寫日志（WAL）和兩階段提交（2PC）兩種方式。

如果外部系統不支持事務，那么可以用預寫日志的方式，把結果數據先當成狀態保存，然后在收到 checkpoint 完成的通知時，一次性寫入 sink 系統。

 

9. Flink常用的算子有哪些

分兩部分：

(1).數據讀取，這是Flink流計算應用的起點，常用算子有：

? 從內存讀：fromElements

? 從文件讀：readTextFile

? Socket 接入 ：socketTextStream

? 自定義讀取：createInput

(2).處理數據的算子，常用的算子包括：Map（單輸入單輸出）、FlatMap（單輸入、多輸出）、Filter（過濾）、KeyBy（分組）、Reduce（聚合）、Window（窗口）、Connect（連接）、Split（分割）等。

 

10. Flink任務延時高，如何入手？

在 Flink 的后臺任務管理中，我們可以看到 Flink 的哪個算子和 task 出現了反壓。

最主要的手段是資源調優和算子調優。

資源調優即是對作業中的 Operator 的并發數（parallelism）、CPU（core）、堆內存（heap_memory）等參數進行調優。

作業參數調優包括：并行度的設置，State 的設置，checkpoint 的設置。

 

11. Flink是如何處理反壓的

Flink 內部是基于 producer-consumer 模型來進行消息傳遞的，Flink的反壓設計也是基于這個模型。Flink 使用了高效有界的分布式阻塞隊列，就像 Java 通用的阻塞隊列（BlockingQueue）一樣。下游消費者消費變慢，上游就會受到阻塞。

 

12. 如何排查生產環境中的反壓問題

反壓出現的場景

反壓經常出現在促銷、熱門活動等場景。短時間內流量陡增造成數據的堆積或者消費速度變慢。

它們有一個共同的特點：數據的消費速度小于數據的生產速度。

反壓監控方法

通過Flink Web UI發現反壓問題。

Flink 的 TaskManager 會每隔 50ms 觸發一次反壓狀態監測，共監測 100 次，并將計算結果反饋給 JobManager，最后由 JobManager 進行計算反壓的比例，然后進行展示。

這個比例展示邏輯如下：

OK: 0 <= Ratio <= 0.10，表示狀態良好正；

LOW: 0.10 < Ratio <= 0.5，表示有待觀察；

HIGH: 0.5 < Ratio <= 1，表示要處理了（增加并行度/subTask/檢查是否有數據傾斜/增加內存）。

0.01，代表100次中有一次阻塞在內部調用。

flink反壓的實現方式

Flink任務的組成由基本的“流”和“算子”構成，“流”中的數據在“算子” 間進行計算和轉換時，會被放入分布式的阻塞隊列中。當消費者的阻塞隊列滿時，則會降低生產者的數據生產速度

反壓問題定位和處理

Flink會因為數據堆積和處理速度變慢導致checkpoint超時，而checkpoint是Flink保證數據一致性的關鍵所在，最終會導致數據的不一致發生。

數據傾斜：可以在 Flink 的后臺管理頁面看到每個 Task 處理數據的大小。當數據傾斜出現時，通常是簡單地使用類似 KeyBy 等分組聚合函數導致的，需要用戶將熱點 Key 進行預處理，降低或者消除熱點 Key 的影。

GC：不合理的設置 TaskManager 的垃圾回收參數會導致嚴重的 GC 問題，我們可以通過 -XX:+PrintGCDetails 參數查看 GC 的日志。

代碼本身：開發者錯誤地使用 Flink 算子，沒有深入了解算子的實現機制導致性能問題。我們可以通過查看運行機器節點的 CPU 和內存情況定位問題。

 

13. Flink中的狀態存儲

Flink在做計算的過程中經常需要存儲中間狀態，來避免數據丟失和狀態恢復。選擇的狀態存儲策略不同，會影響狀態持久化如何和 checkpoint 交互。

Flink 提供了三種狀態存儲方式：MemoryStateBackend、FsStateBackend、RocksDBStateBackend。

 

14. Operator Chains（算子鏈）這個概念你了解嗎

為了更高效地分布式執行，Flink 會盡可能地將 operator 的 subtask 鏈接（chain）在一起形成 task，每個 task 在一個線程中執行。

將 operators 鏈接成 task 是非常有效的優化：它能減少線程之間的切換，減少消息的序列化/ 反序列化，減少數據在緩沖區的交換，減少了延遲的同時提高整體的吞吐量。這就是我們所說的算子鏈。

 

15. Flink的內存管理是如何做的

Flink 并不是將大量對象存在堆上，而是將對象都序列化到一個預分配的內存塊上。此外，Flink大量的使用了堆外內存。如果需要處理的數據超出了內存限制，則會將部分數據存儲到硬盤上。Flink 為了直接操作二進制數據實現了自己的序列化框架。

 

16. 如何處理生產環境中的數據傾斜問題

(1).Flink數據傾斜的表現：

任務節點頻繁出現反壓，增加并行度也不能解決問題；

部分節點出現OOM異常，是因為大量的數據集中在某個節點上，導致該節點內存被爆，任務失敗重啟。

(2).數據傾斜產生的原因：

業務上有嚴重的數據熱點，比如滴滴打車的訂單數據中北京、上海等幾個城市的訂單量遠遠超過其他地區；

技術上大量使用了 KeyBy、GroupBy 等操作，錯誤的使用了分組 Key，人為產生數據熱點。

(3).解決問題的思路：

業務上要盡量避免熱點 key 的設計，例如我們可以把北京、上海等熱點城市分成不同的區域，并進行單獨處理；

技術上出現熱點時，要調整方案打散原來的 key，避免直接聚合；此外 Flink 還提供了大量的功能可以避免數據傾斜。

 

17. Flink中的Time有哪幾種？

（1）、Flink中的事件時間（Event Time）和處理時間（Processing Time）有什么區別？

事件時間是數據實際生成的時間，而處理時間是數據到達Flink系統的時間。事件時間可以通過時間戳標記數據，而處理時間是Flink根據數據到達的順序生成的。

 

（2）、Flink中的時間有三種類型: 

EventTime：是事件創建的時間。它通常由事件中的時間戳描述，例如采集的日志數據中，每一條日志都會記錄自己的生成時間，Flink通過時間戳分配器訪問事件時間戳。

Ingestion Time：是數據進入Flink的時間。

Processing Time：是每一個執行基于時間操作的算子的本地系統時間，與機器相關，默認的時間屬性就是Processing Time。

例如，一條日志進入Flink的時間為2021-01-2210:00:00.123，到達Window的系統時間為2021-01-22 10:00:01.234，日志的內容如下：

2021-01-06 18:37:15.624 INFO Fail over to rm2

對于業務來說，要統計1min內的故障日志個數，哪個時間是最有意義的？—— eventTime，因為我們要根據日志的生成時間進行統計。

 

18. Flink對于遲到數據是怎么處理的

Flink中 WaterMark 和 Window 機制解決了流式數據的亂序問題，對于因為延遲而順序有誤的數據，可以根據eventTime進行業務處理，對于延遲的數據Flink 也有自己的解決辦法，主要的辦法是給定一個允許延遲的時間，在該時間范圍內仍可以接受處理延遲數據

設置允許延遲的時間是通過allowedLateness(lateness: Time)設置

保存延遲數據則是通過sideOutputLateData(outputTag: OutputTag[T])保存

獲取延遲數據是通過DataStream.getSideOutput(tag: OutputTag[X])獲取

文章推薦：

Flink 中極其重要的 Time 與 Window 詳細解析

 

19. Flink中window出現數據傾斜怎么解決

window 產生數據傾斜指的是數據在不同的窗口內堆積的數據量相差過多。本質上產生這種情況的原因是數據源頭發送的數據量速度不同導致的。出現這種情況一般通過兩種方式來解決：

 

在數據進入窗口前做預聚合

重新設計窗口聚合的 key

 

20. Flink CEP編程中當狀態沒有到達的時候會將數據保存在哪里

在流式處理中，CEP 當然是要支持 EventTime 的，那么相對應的也要支持數據的遲到現象，也就是watermark的處理邏輯。CEP對未匹配成功的事件序列的處理，和遲到數據是類似的。在 Flink CEP的處理邏輯中，狀態沒有滿足的和遲到的數據，都會存儲在一個Map數據結構中，也就是說，如果我們限定判斷事件序列的時長為5分鐘，那么內存中就會存儲5分鐘的數據，這在我看來，也是對內存的極大損傷之一。

 

21. Flink設置并行度的方式

們在實際生產環境中可以從四個不同層面設置并行度：

(1)、操作算子層面(Operator Level)

.map(new RollingAdditionMapper()).setParallelism(10) //將操作算子設置并行度

(2)、執行環境層面(Execution Environment Level)

$FLINK_HOME/bin/flink 的-p參數修改并行度

(3)、客戶端層面(Client Level)

env.setParallelism(10)

(4)、系統層面(System Level)

全局配置在flink-conf.yaml文件中，parallelism.default，默認是1：可以設置默認值大一點。

需要注意的優先級：算子層面>環境層面>客戶端層面>系統層面。

 

22. Flink中Task如何做到數據交換

在一個 Flink Job 中，數據需要在不同的 task 中進行交換，整個數據交換是有 TaskManager 負責的，TaskManager 的網絡組件首先從緩沖 buffer 中收集 records，然后再發送。Records 并不是一個一個被發送的，是積累一個批次再發送，batch 技術可以更加高效的利用網絡資源。

 

23. Flink的內存管理是如何做的

Flink 并不是將大量對象存在堆上，而是將對象都序列化到一個預分配的內存塊上。此外，Flink大量的使用了堆外內存。如果需要處理的數據超出了內存限制，則會將部分數據存儲到硬盤上。Flink 為了直接操作二進制數據實現了自己的序列化框架。

 

24. 介紹下Flink的序列化

Flink 摒棄了 Java 原生的序列化方法，以獨特的方式處理數據類型和序列化，包含自己的類型描述符，泛型類型提取和類型序列化框架。

TypeInformation 是所有類型描述符的基類。它揭示了該類型的一些基本屬性，并且可以生成序列化器。

TypeInformation 支持以下幾種類型：

BasicTypeInfo: 任意 Java 基本類型或 String 類型

BasicArrayTypeInfo: 任意 Java 基本類型數組或 String 數組

WritableTypeInfo: 任意 Hadoop Writable 接口的實現類

TupleTypeInfo: 任意的 Flink Tuple 類型(支持 Tuple1 to Tuple25)。

Flink tuples 是固定長度固定類型的 Java Tuple 實現

CaseClassTypeInfo: 任意的 Scala CaseClass(包括 Scala tuples)

PojoTypeInfo: 任意的 POJO (Java or Scala)，例如，Java 對象的所有成員變量，要么是 public 修飾符定義，要么有 getter/setter 方法

GenericTypeInfo: 任意無法匹配之前幾種類型的類

 

25. Flink海量數據高效去重

基于狀態后端。

基于HyperLogLog：不是精準的去重。

基于布隆過濾器（BloomFilter）: 快速判斷一個key是否存在于某容器，不存在就直接返回。

基于BitMap: 用一個bit位來標記某個元素對應的Value，而Key即是該元素。由于采用了Bit為單位來存儲數據，因此可以大大節省存儲空間。

基于外部數據庫:選擇使用Redis或者HBase存儲數據，我們只需要設計好存儲的Key即可，不需要關心Flink任務重啟造成的狀態丟失問題。

 

26. Flink SQL的是如何實現的

構建抽象語法樹的事情交給了 Calcite 去做。SQLquery 會經過 Calcite 解析器轉變成 SQL 節點樹，通過驗證后構建成 Calcite 的抽象語法樹（也就是圖中的 Logical Plan）。另一邊，Table API 上的調用會構建成 Table API 的抽象語法樹，并通過 Calcite 提供的 RelBuilder 轉變成 Calcite 的抽象語法樹。

 

然后依次被轉換成邏輯執行計劃和物理執行計劃。

在提交任務后會分發到各個 TaskManager 中運行，在運行時會使用 Janino 編譯器編譯代碼后運行。

 

27.FLink的窗口

（1）、Flink的窗口（Window）
窗口是Flink中用于對無限數據流進行有界處理的機制。它將無限流切分為有限的、不重疊的塊，并對每個窗口進行計算。
（2）、支持窗口類型
Flink支持滾動窗口（Tumbling Window）、滑動窗口（Sliding Window）、會話窗口（Session Window）。