?摘要：本文整理自快手實時計算團隊技術(shù)專家劉建剛在 Flink Forward Asia 2021 生產(chǎn)實踐專場的演講。主要內(nèi)容包括：

1. 快手 Flink 的歷史及現(xiàn)狀
2. Flink 容錯能力提升
3. Flink 引擎控制與實踐
4. 快手批處理實踐
5. 未來規(guī)劃

01快手 Flink 的歷史與現(xiàn)狀

快手從 2018 年開始對 Flink 進行深度整合，經(jīng)過 4 年發(fā)展，實時計算平臺逐漸完善并賦能周邊各種組件。

• 2018 年我們針對 Flink 1.4 進行了平臺化建設(shè)并大幅提升運維管理能力，達到了生產(chǎn)可用。
• 2019 年我們開始基于 1.6 版本進行迭代開發(fā)，很多業(yè)務都開始實時化，比如優(yōu)化 interval join 為商業(yè)化等平臺帶來顯著收益、開發(fā)實時多維分析加速超大多維報表的實時化，這一年我們的 Flink SQL 平臺也投入使用。
• 到了 2020 年，我們升級到 1.10，對 sql 的功能進行了非常多的完善，同時進一步優(yōu)化 Flink 的核心引擎，保障了 Flink 的易用性、穩(wěn)定性、可維護性。
• 2021 年我們開始發(fā)力離線計算，支持湖倉一體的建設(shè)，進一步完善 Flink 生態(tài)。

上圖是快手基于 Flink 的技術(shù)棧。

• 最核心、最底層是 Flink 的計算引擎，包括流計算和批處理，我們針對穩(wěn)定性和性能做了大量工作。
• 外面一層是跟 Flink 打交道的周邊組件，既有 Kafka、rocketMQ 等中間件，也有 ClickHouse、Hive 等數(shù)據(jù)分析工具，還有 Hudi 等數(shù)據(jù)湖的使用。用戶可以基于 Flink 和這些組件構(gòu)建各種應用，涵蓋了實時、近實時、批處理的各種場景。
• 最外層是具體的使用場景，常見的有電商、商業(yè)化等視頻相關(guān)的業(yè)務方，應用場景包含機器學習、多維分析等。另外還有很多技術(shù)部門基于 Flink 來實現(xiàn)數(shù)據(jù)的導入、轉(zhuǎn)換，比如 CDC、湖倉一體等。

應用規(guī)模上，我們有 50 萬 CPU 核，主要通過 Yarn 和 K8s 的方式進行資源托管，上面運行著 2000+ 作業(yè)，峰值處理達到了 6億/秒，日處理條數(shù)達到了 31.7 萬億，節(jié)假日或活動的時候流量甚至會翻倍。

02容錯能力提升

容錯能力主要包含以下部分：

• 首先是單點恢復，支持任意多個 task 失敗時的原地重啟，long-running 作業(yè)基本可以做到永不斷流；
• 其次，是集群故障的應對，包含冷備、熱備以及 Kafka 雙集群的集成；最后是黑名單的使用。

Flink 為了做到 exactly-once，任何節(jié)點出現(xiàn)故障都需要重啟整個作業(yè)，全局重啟會帶來長時間的停頓，最高可達十幾分鐘。有些場景不追求 exactly-once，比如推薦等實時場景，但它們對服務可用性的要求很高，無法容忍作業(yè)的斷流，還有模型訓練等初始化很慢的場景，重啟時間特別長，一旦重啟將會造成很大的影響?；谝陨峡紤]，我們開發(fā)了單點恢復功能。

上圖是單點恢復的基本原理。如圖有三個 task，其中中間的 task 失敗了，那么首先 Flink 的主節(jié)點會重新調(diào)度中間的 task，此時上下游的 task 不會失敗，而是等待重連。等中間的 task 調(diào)度成功后，master 節(jié)點會通知下游的 task 去重連上游的 task，與此同時中間的 task 也會去連它上游的 task，通過重新構(gòu)建讀視圖來恢復數(shù)據(jù)的讀取。等上下游都連接成功后這個作業(yè)就可以正常工作了。

了解完基本原理，再來看一下線上多 task 恢復的案例。實際環(huán)境中經(jīng)常會出現(xiàn)多個 task 同時失敗的情況，這個時候我們會按照拓撲順序來逐個恢復失敗的 task，比如上圖中是按照從左往右的順序恢復。

這個功能上線之后，我們內(nèi)部有將近 100 個作業(yè)使用了這個功能，常規(guī)故障下作業(yè)都可以做到不斷流，即便出現(xiàn)小的流量波動，業(yè)務也可以做到無感知，業(yè)務方徹底告別了服務斷流的噩夢。

集群故障一旦發(fā)生就是致命性的，所有的數(shù)據(jù)都會流失，服務也會掛掉。我們的方案主要包含冷備、熱備，以及 Flink 和 Kafka 的雙集群集成。

冷備主要指的是對數(shù)據(jù)做備份，集群掛掉以后可以快速在另外一個集群啟動計算任務。

如上圖，KwaiJobManager 是快手的作業(yè)管理服務，其中的 failover coordinator 主要負責故障處理。我們會把所有 jar 包等文件保存在 HDFS，所有的信息保存在 Mysql，這兩者都做到了高可用。作業(yè)運行在主集群 ClusterA，線上用的是增量快照，會存在文件依賴的問題，所以我們定期做 savepoint 并拷貝到備集群。為了避免文件過多，我們設(shè)置了定時刪除歷史快照。

一旦服務檢測到集群 A 故障，就會立刻在集群B啟動作業(yè)，并從最近一次的快照恢復，確保了狀態(tài)不丟失。對于用戶來說，只需要設(shè)置一下主備集群，剩下的全都交由平臺方來做，用戶全程對故障無感知。

熱備就是雙集群同時運行一樣的任務。我們的熱備都是全鏈路的，Kafka 或者 ClickHouse 等都是雙跑。最上面的展示層只會使用其中一份結(jié)果數(shù)據(jù)做展示，一旦出現(xiàn)故障，展示層會立刻切換到另外一份數(shù)據(jù)，切換過程在一秒以內(nèi)，用戶全程無感知。

相比冷備，熱備需要等量的資源來備份運行，但切換的速度更快，比較適用于春晚等要求極高的場景。

Flink 與 Kafka 的雙集群集成，主要是因為快手的 Kafka 都具備雙集群的能力，所以需要 Flink 支持讀寫雙集群的 Kafka topic，這樣某個 Kafka 集群掛掉時Flink可以在線無縫切換。如上圖所示，我們 Flink 對 Kafka 雙集群做了抽象，一個邏輯上的 topic 底層對應兩個物理上的 topic，里面由多個 partition 組合而成，F(xiàn)link 消費邏輯 topic 就相當于同時讀取底層兩個物理 topic 的數(shù)據(jù)。

針對集群的各種變動，我們?nèi)砍橄蟪闪?partition 上的擴縮容，比如集群掛掉，可以看成是邏輯 topic 的 partition 縮容；單集群切雙集群，可以看成是邏輯 topic 的擴容；topic 的遷移，可以看成邏輯 topic 先擴容再縮容。這里我們都是按照雙集群來舉例，實際上無論是雙集群還是更多的集群，原理都是一樣的，我們都提供了支持。

出現(xiàn)以下兩種情況的時候需要使用黑名單功能。第一種是反復調(diào)度有故障的機器，導致作業(yè)頻繁失敗。另一種是機器因為硬件或網(wǎng)絡(luò)等原因，導致 Flink 個別節(jié)點卡主但未失敗。

針對第一種情況，我們開發(fā)了閾值拉黑，如果作業(yè)在同一個機器上失敗或者多次部署閾值失敗，超過配置的閾值就會拉黑；針對第二種情況，我們建立了異常分類機制，針對網(wǎng)絡(luò)卡頓和磁盤卡頓情況，直接驅(qū)除容器并且拉黑機器。另外我們還對外暴露了拉黑接口，打通了運維 Yarn 等外部系統(tǒng)，實現(xiàn)了實時拉黑。我們還以 Flink 黑名單為契機，建立了一套完整的硬件異常處理流程，實現(xiàn)了作業(yè)自動遷移，全程自動化運維，用戶無感知。

03Flink 引擎控制與實踐

3.1 Flink實時控制?

針對 long-running 的實時作業(yè)，用戶經(jīng)常需要作出變更比如調(diào)整參數(shù)來更改行為，還有一些系統(tǒng)運維比如作業(yè)降級、修改日志級別等，這些變更都需要重啟作業(yè)來生效，有時會高達幾分鐘到幾十分鐘，在一些重要的場合，這是無法容忍的。比如在活動期間或者排查問題的關(guān)鍵點，作業(yè)一旦停止將會功虧一簣，所以我們需要在不停止作業(yè)的情況下實時調(diào)整作業(yè)的行為，也就是實時控制。

從更廣泛的角度來看，F(xiàn)link 不僅是計算任務，也是一個 long-running service。我們的實時控制正是基于這樣的考慮，來為實時計算提供交互式的控制模式。如上圖所示，用戶通過經(jīng)典的 kv 數(shù)據(jù)類型與 Flink dispatcher 交互，F(xiàn)link 收到消息后，會先將它們持久化到 zk 用于 failover，然后根據(jù)具體的消息做相應的控制，比如控制 resource manager、控制 job master 或者其他組件。

我們既支持用戶自定義動態(tài)參數(shù)，也為用戶提供了很多現(xiàn)成的系統(tǒng)控制。用戶自定義主要是使用 RichFunction 來獲取動態(tài)參數(shù)，并且實現(xiàn)相應的邏輯，這樣在作業(yè)運行的時候就可以實時傳入?yún)?shù)，達到實時控制的效果。

系統(tǒng)提供的實時控制能力，主要包含數(shù)據(jù)源限速、采樣、重置 Kafka offset、調(diào)整快照參數(shù)以及運維相關(guān)的更改日志級別、拉黑節(jié)點等功能。除此之外，我們還支持動態(tài)修改部分 Flink 原生配置。

快手內(nèi)部對實時控制功能實現(xiàn)了產(chǎn)品化，用戶使用起來非常方便。

3.2 源端控制能力?

Flink 處理歷史任務或者作業(yè)性能跟不上的的時候，會引發(fā)以下的問題:

首先 source 的各個并發(fā)處理速度不一致，會進一步加重數(shù)據(jù)的亂序、丟失、對齊慢等問題。其次，快照會持續(xù)變大，嚴重影響作業(yè)性能。另外還有流量資源的不可控，在高負載的情況下會引發(fā) CPU 打滿、oom 等穩(wěn)定性問題。

由于 Flink 是一種 pipeline 實時計算，因此從數(shù)據(jù)源入手可以從根本上解決問題。

首先來看下歷史數(shù)據(jù)精準回放功能。上圖是以二倍速率去消費 Kafka 的歷史數(shù)據(jù)，F(xiàn)link 作業(yè)追上 lag 之后就可以轉(zhuǎn)成實時消費。通過這種方式可以有效解決復雜任務的穩(wěn)定性問題。

上圖的公式是一個基本原理，消費倍率 = Kafka 的時間差 / Flink 的系統(tǒng)時間差，用戶使用的時候只需要配置倍率即可。

另外一個能力是 QPS 限速。數(shù)據(jù)流量很大的時候，會導致 Flink 的負載很高以及作業(yè)不穩(wěn)定。我們基于令牌桶算法，實現(xiàn)了一套分布式的限速策略，可以有效減緩 Flink 的壓力。使用 QPS 限速后，作業(yè)變得非常健康，上圖綠色部分可見。19 年的春晚大屏，我們就是通過這個技術(shù)實現(xiàn)了柔性可用的保障。

另外我們還支持自動適配 partition 的變更和實時控制，用戶可以隨時隨地調(diào)整作業(yè)的 QPS。

最后一個功能是數(shù)據(jù)源對齊，主要指 watermark 的對齊。首先每個 subtask 都會定期向主節(jié)點匯報自己的 watermark 進度，主要包括 watermark 的大小和速度。主節(jié)點會計算下一個周期的 target，即預期的最大 watermark，再加一個 diff 返回給各個節(jié)點。各個 source task 會保證下一個周期的 watermark 不超過設(shè)置的 target。上圖最下面是 target 的計算公式，預測每個 task 下個周期結(jié)束時候的 waterMark 值，再加上我們允許的 maxdiff 然后取最大值，通過這種方式可以保障各個 source 的進度一致，避免 diff 過大導致的穩(wěn)定性問題。

3.3 作業(yè)均衡調(diào)度

生產(chǎn)環(huán)境中經(jīng)常會出現(xiàn)資源不均衡的現(xiàn)象，比如第一點 Flink 的 task 分布不均勻，導致 task manager 資源使用不均衡，而作業(yè)的性能又往往受限于最繁忙的節(jié)點。針對這個問題，我們開發(fā)了作業(yè)均衡調(diào)度的策略；第二點是 CPU 使用不均衡，有些機器被打滿而有些機器很閑。針對這個問題，我們開發(fā)了 CPU 均衡調(diào)度的功能。

上圖中有三個 jobVertex，通過 hash shuffle 的方式來連接。上圖中間部分顯示了 Flink 的調(diào)度，每個 jobVertex 都是自上而下往 slot 里調(diào)度 task，結(jié)果就是前兩個 slot 很滿而其他 slot 很空閑，第一個 task manager 很滿而第二個 task manager 很空閑。這是一個很典型的資源傾斜的場景，我們對此進行了優(yōu)化。調(diào)度的時候首先計算需要的總資源，也就是需要多少個 task manager，然后計算每個 TM 分配的 slot 個數(shù)，確保 TM 中的 slot 資源均衡。最后均衡分配 task 到各個 slot 中，確保 slot 中 task 均衡。

實際運行過程中還存在另外一種傾斜情況 —— CPU 傾斜，我們來看下怎么解決這個問題。上圖左側(cè)，用戶申請了一個核但實際只使用了 0.5 個核，也有申請了一個核實際使用了一個核。按照默認調(diào)度策略，大量此類 case 可能會導致有的機器 CPU 使用率很高，有的卻很閑，負載高的機器不論是性能還是穩(wěn)定性都會比較差。那么如何讓申請和使用的 diff 盡可能?。?/p>

我們的方案是對作業(yè)資源精準畫像，具體做法分為以下步驟：作業(yè)運行過程中統(tǒng)計每個 task 所在容器的 CPU 使用率，然后建立 task 到 executionSlotSharingGroup，再到 container 的映射，這樣就知道了每個 task 所在 slot 的 CPU 使用情況，然后根據(jù)映射關(guān)系重啟作業(yè)，根據(jù) task 所在 slot 的歷史 CPU 使用率來申請相應的資源，一般來說會預留一些 buffer。如上圖右圖所示，如果預測足夠準，重啟后 task manager 使用的資源不變，但是申請值變小了，二者的 diff 就變小了。

其實業(yè)界一些先進的系統(tǒng)，比如 borg 是支持動態(tài)修改申請值的，但我們的底層調(diào)度資源不持這種策略，所以只能在 Flink 這一層使用資源畫像來解決這個問題。當然資源畫像不能保證百分百準確，我們還有其他策略，比如限制高 CPU 負載的機器繼續(xù)分配資源，盡可能減少不均衡。另外我們還建立了分級保障制度，不同優(yōu)先級的作業(yè)有不同的 cgroup 限制，比如低優(yōu)先級作業(yè)不再超配，高優(yōu)先級作業(yè)允許少量超配，從而避免 CPU 使用過多導致的不均衡。

04快手批處理實踐

上圖是我們的批處理架構(gòu)圖。最底層為離線集群，中間是 Flink 引擎以及 Flink 的 data stream API、SQL API，再上面是一些平臺方比如 sql 入口、定時調(diào)度平臺等，此外還有一些流批一體的探索，最上面是各種用戶比如視頻、商業(yè)化等。

流批一體中，流的特性是低延時，批的特性是高吞吐。針對流批一體，我們期待系統(tǒng)既能處理 unfield batch 數(shù)據(jù)，也可以調(diào)整數(shù)據(jù)塊的 shuffle 大小等來均衡作業(yè)的吞吐和時延。

快手內(nèi)部對流批一體進行了很多探索，我們?yōu)榇鎯?shù)據(jù)建立了統(tǒng)一的 Schema 標準，包括流表和批表，用戶可以使用相同的代碼來處理流表和批表，只是配置不同。產(chǎn)生的結(jié)果也需要符合統(tǒng)一的 Schema 標準，這樣就可以打通上下游，實現(xiàn)盡可能多的邏輯復用。Schema 統(tǒng)一是我們快手數(shù)據(jù)治理的一部分，湖倉一體等場景也都有這個需求。

應用場景主要包括以下幾個方面：

• 指標計算，比如實時指標和報表計算。
• 數(shù)據(jù)回溯，利用已有的離線數(shù)據(jù)重新生成其他指標。
• 數(shù)倉加速，主要是數(shù)據(jù)倉庫和數(shù)據(jù)湖的實時加速。

流批一體帶來的收益是多方面的，首先是降低了開發(fā)和運維成本，實現(xiàn)了盡可能多的代碼邏輯復用，運維不再需要維護多個系統(tǒng)。其次是實時處理和批處理的口徑保持一致，保障了最終結(jié)果的一致。最后是資源方面的收益，有些場景只需要一套實時系統(tǒng)。

我們在調(diào)度方面進行了優(yōu)化。如上圖所示的三個 task，起初 a 和 c 已經(jīng)完成，b 還在運行。這時 a 失敗了，按照默認的策略 ABC 都需要重新運行，即便 c 已經(jīng)完成。在實際場景中會有大量的 c 進行重算，帶來巨大的資源損耗。針對這種情況如果，我們默認開啟了以下策略：如果 a 的結(jié)果是決定性的（實際上大部分批處理的輸出都是決定性的），可以不再重算 c，只需計算 a 和 b。

上圖是我們快手內(nèi)部針對批處理的優(yōu)化和改進。

第一個是 shuffle service，現(xiàn)在既有內(nèi)部的集成，也在試用社區(qū)的版本，主要是為了實現(xiàn)存儲和計算的解耦，同時提高 shuffle 的性能。第二個是動態(tài)資源的調(diào)度，主要是根據(jù)數(shù)據(jù)量來自動決定算子的并發(fā)，避免人工反復調(diào)整。第三個是慢節(jié)點規(guī)避，也叫推測執(zhí)行，主要是為了減少長尾效應，減少總執(zhí)行時間。第四個是 hive 的優(yōu)化，比如 UDF 適配、語法兼容。另外針對 partition 生成 split，我們增加了緩存、多線程生成等方式，極大減少了分片的時間。最后是一些壓縮方式的支持，比如支持 gzip、zstd 等。

05未來規(guī)劃

我們的未來規(guī)劃主要分為以下幾個方面：

• 首先是實時計算，進一步增強 Flink 的性能、穩(wěn)定性和應用性，并通過實時計算來加速各種業(yè)務場景。
• 第二個是在線和離線的統(tǒng)一，包含實時、近實時和批處理。我們期待能用 Flink 統(tǒng)一快手的數(shù)據(jù)同步、轉(zhuǎn)換和在離線計算，讓ETL、數(shù)倉、數(shù)據(jù)湖處理等各類場景，都使用一套 Flink 計算系統(tǒng)。
• 最后一個是彈性可伸縮，主要是云原生相關(guān)，包含在離線混部和作業(yè)的彈性伸縮等。?