Apache Flink 在快手的過去、現在和未來

本文由快手大資料架構團隊負責人趙健博分享，主要介紹 Apache Flink 在快手的過去、現在和未來。内容包括：

1. 為什麼選 Flink
2. Flink 在快手的發展
3. 業務資料流
4. 技術創新
5. 未來計劃

一、為什麼選 Flink

大家好，我是趙健博，來自快手，目前負責快手大資料架構團隊。今天很高興可以和大家分享我們在 Flink 項目上的應用、改進與發展曆程。

先來看一下我們選擇 Flink 引擎的主要原因：

• 首先，Flink 能做到亞秒級處理延遲。目前大部分的業務需求對實時處理延遲要求越來越高，這是個最基本需求。
• 其次，Flink 有豐富的視窗計算模式，且自帶狀态存儲引擎以及精準一次的語義，這個能力極大簡化了資料的處理複雜度，顯著提升了研發的速度。
• 最後，批流一體能力以及研發模式的變革，也将進一步提效研發，為業務賦能。

本次會議也看到了很多公司都在分享批流一體落地實踐，相信流批一體全場景落地的大程序也将指日可待。

二、Flink 在快手的發展

Flink 在快手的發展曆程，總的來說可以分為四個階段：

1. 我們是從 17 年開始使用 Flink 的，17 年我們主要是初步試用，當時接入的業務是直播與短視訊的品質監控業務。
2. 進入到 2018 年之後，在能力上，我們開始對 Flink 進行成周邊體系的建設，例如，建構引擎内部 metric 的采集，監控與報警流程、作業托管平台上線等。與此同時，我們也在不斷的加深對 Flink 的了解，修煉内功；在業務上，開始接入直播 CDN 流量排程，日志實時拆分、投放分析、用戶端 Crash 分析等場景。
3. 進入到 2019 年後，随着對 Flink 引擎掌控力的加強，我們開始進行一些穩定性與性能相關的改進，主要包括防雪崩，流控、分級保障、參數熱更新、自研狀态存儲引擎 Slimbase、實時多元模組化等。在業務上，開始支撐春節活動大屏、實時多元分析、曝光/點選流實時 Join 等場景。
4. 到 2020 年後，我們除了持續關注穩定性性能之外，也在推進效率改進，例如調研并開始試用 Flink SQL，以及流批一體能力。在業務上，采用 Flink SQL 支撐活動大屏、開始通過 Flink 以及流批一體能力建設 AI 資料流、實時報表、直播精彩時刻等業務場景。

截止到目前，快手 Flink 從業務規模上看有若幹叢集，叢集有數千機器，目前還是部署在 YARN 上，後續也會考慮遷移到 K8s 上。總的作業 2000 左右，這些作業每天處理 20 多萬億條的記錄，其中峰值達到每秒 6 億條的規模。

三、業務資料流

1. 資料流的總體架構圖

接下來，讓我們看下快手 Flink 目前應用的一些業務場景與業務資料流的案例。

下面這張圖是一個資料流的總體架構圖，從這張圖中，大家能看到資料的源頭有三類資料，一個是資料庫中的資料，一個是服務端的日志，最後是用戶端的日志，這些日志上報給 Kafka 的服務。

在快手，所有日志或者消息都是通過 Kafka 服務流轉的。資料進入到 Kafka 之後分别流轉到實時資料鍊路，以及離線資料鍊路上（實時同步到 Hive）。在實時鍊路上，目前 Flink 支撐了很多業務場景，如：實時 ETL、資料內建、實時報表計算、實時監控、實時實時特征等等。這些資料通過 Flink 實時計算處理之後，将流入到各種類型的資料庫中，例如多元資料庫（Druid/Clickhouse），MySQL、Redis、HBase 等等。之後各類的資料産品、資料應用、業務應用從這些資料庫中擷取最新的聚合或者結果資料，進行業務的處理。

2. 實時 ETL 場景

接下來，我們展開介紹下上述各個場景下的業務資料流圖。在實時 ETL 場景下，目前我們主要在推廣使用 Flink SQL 進行資料的實時 ETL。下圖左側展示了實時 ETL的流程，其中 Kafka 中的 topic 的 schema 都被中繼資料服務管理起來了。Flink 引擎首先通路中繼資料中心擷取 Topic 的 schema，然後将 Topic 轉成實時表，并通過 SQL 完成 ETL 的處理落地。右側的 SQL 是我們進行資料拆分的案例。

3. 資料內建場景

在資料內建場景下，如左圖所示，通過 Flink 引擎可以很友善地完成 Kafka/HBase/ES/Hive/Redis 等服務的資料交換。相比于其他引擎，Flink 的 source/sink 支援的服務種類更豐富，且更加友善擴充。除此之外，除了離線資料交換，Flink 是天然支撐實時場景的。

4. 實時報表的場景

在實時報表的場景下，介紹下 Flink 支援快手春節活動的實時資料鍊路。

如圖所示，整個資料流從左到右共分為 4 層，分别是 ODS 層、采樣層、名額邏輯計算層、資料服務層。

1. 最開始是原始的 ODS 層資料，通過用戶端，服務端，或者是 DB 直接打到 Kafka 的 topic 中形成一個 ODS 層，這一層的資料經過 Flink 的處理，再寫回 Kafka，形成一個采樣層。
2. 采樣層提出來的原因主要是，面向春節活動的流量高峰，沒法精準預知它的峰值有多高，是以我們需要具備對整個流量進行采樣的能力，以便能夠在有限的資源下應對洪峰。一旦洪峰來了，可以進行資料采樣處理，有效降低計算資源的消耗，同時再通過采樣的規則在後續邏輯計算層還原采樣之前資料名額的結果。
3. 資料被采樣之後再通過 Flink 進行邏輯層的計算，例如留存、新增、PV、UV 等名額，然後将這些名額最終儲存到 Redis 或者多元引擎中。在這個計算過程中，當時采用的是外部存儲與服務進行了 UV，以及新增的計算。在未來的活動支撐中，我們會逐漸替換為 Flink 自己的 state 引擎。
4. 最後，各類資料産品與服務，如大屏，看闆等，從 Redis 或者多元引擎中擷取資料進行展示以及政策的調整。

1. 實時監控場景

在實時監控這個場景下，介紹下快手直播品質監控和 CDN 流量排程鍊路。

首先資料通過埋點采集，打到 Kafka 之後，在實時鍊路的處理上，通過 Flink 進行資料的清洗、轉換、聚合，形成 DWD 和 DWS 層的資料，這些資料也會最終寫回 Kafka。之後，會把 DWS 層的資料導到後面的 OLAP 這種資料庫中。然後上層的 BI 服務通過通路資料庫中的資料進行報表的展示，進而完成監控，以及資料決策。同時，品質計算的排程結果資料存儲 Redis，供線上 CDN 排程服務提供決策依據。

在離線鍊路上，可以考慮從實時鍊路中的每一層進行資料的導出，導到 Hive 表中。這部分資料的儲存主要是為了解決 Ad hoc 分析，以及當實時流資料出現問題，進行的離線的資料修正。

6. 特征處理場景

最後一個業務場景，介紹一個 AI 資料流案例，特征處理與索引生成流程。在快手，有大量的特征需要處理，特征的處理與管理效率對模型疊代效率有很大影響。采用 Flink 進行特征與索引的處理，在管理上與研發效率上都有比較大的優勢。

目前我們借助 Flink 完成了一部分的特征與索引生産流程，如圖所示，行為資料通過 Kafka 流入 Flink 之後，利用 Flink 的視窗計算能力完成各種類型的特征實時計算，之後将特征存儲到特征庫中，同時也會同步一份資料到 Hive 中，用作做特征離線資料流處理；除此之外，當有索引需要生成的時候，會通過 Kafka 觸發生成政策，下遊的索引生成的 Flink 作業從各種特征庫中擷取特征并進行處理後，形成索引，存儲到索引庫中。最終的索引資料，為線上的推薦服務提供召回源。

四、技術創新

1. 狀态引擎

接下來重點介紹一下 Flink 在快手做的一些技術改進和創新。首先介紹下我們自研的狀态引擎 Slimbase。它在設計上分了三層：

• 接口層，在接口層主要相容目前狀态存儲的幾類接口，value、list、map 狀态等。
• 中間層，我們建構了一個 KV 的 cache 層，主要是做資料的讀和寫的加速。在這層内部，又分為高速 KV 層和 Chunk 層，高速 KV 層（HashMap）有非常快的存取速度，但是空間使用率比較低。為了節省空間，我們又在整個高速 KV 層下面建了一個 Chunk 層，一個 Chunk 是多個 KV 序列化組成的。通過這種序列化的組織之後，在某些場景下相比于 KV 層能夠節省約 60% 的空間。但是在存取速度上會有一定程度的降低。實際使用的時候，可以根據實際情況靈活控制高速 KV 層與 Chunk 層的容量配比。
• 分布式檔案系統層，緩存層被淘汰的資料将會寫入到檔案系統層，最終形成一個個檔案。為了提高檔案系統層面的讀取性能，多個檔案會通過 compaction 進行合并。此外，檔案系統層有檔案塊級别的緩存，具備緩存熱點資料能力。

以上就是 Slimbase 整體架構。下面我們看看 Flink Benchmark 跑出來結果（和RocksDB 對比）。本次測試采用了相同大小的緩存，資料集采用了50w、1500w、5000w 三種規模。

目标是測試三種場景下的結果：

• 僅覆寫高速 KV 緩存；
• 覆寫高速 KV 緩存 +Chunk 緩存；
• 覆寫 KV 高速緩存 +Chunk 緩存+檔案系統；

這是 50 萬的資料集，這些資料集全部是在高速的 KV 層中的。從測試結果上看，相比 RocksDB，Slimbase 讀寫有 3~9 倍的性能提升。

1

在 1500w 資料規模下，資料會分布在高速的 KV 層加 Chunk 層，相比 RocksDB，讀寫有 2~6 倍的性能提升。

在 5000w 資料規模下，資料命中的層次變得更多，把檔案系統也覆寫到了。相比前兩個場景，我們發現性能有比較大的下降。相比 RocksDB，讀性能 0.5~0.7；寫性能 0.90~4 倍。是以我們接下來會在整個檔案系統層的存取性能上，會做專項的優化，提升整個檔案系統的性能，最終可以超過 RocksDB 性能。

2. 穩定性

在介紹穩定性的改進前，我們先來看一下影響 Flink 穩定性的因素有哪些。我這裡總結了三點：

• 硬體故障，例如機器故障，機櫃故障，Tor 故障，機房故障等。
• Flink 依賴的服務異常，例如 Kafka 叢集異常，HDFS 服務異常等。
• Flink 流量過載，例如硬體滿載，以及由于資料源消費速度差異導緻的滿載。

在硬體故障場景下，這裡面取了一個單點的場景。看下這個 Flink 作業，由兩個 source，一個 window 組成。右側是 Flink 作業的實體部署的情況。最大的框代表一台機器，大框裡面的多個小框代表多個 TaskManager。

如果出現了一個節點故障，比如 node3 發生故障了。Flink 引擎會重新從 YARN 申請資源，完成 TaskManager 初始化，并重新部署作業。

我們對一個業務作業做了一個分析，發現當機故障後到作業恢複，共需要 90s 的時間。當機檢測 (60秒)，重新申請資源容器 (5秒)，容器初始化 (20秒)，作業重新部署執行 (5秒）。這對于某些線上業務場景來說是不能接受的。從具體的過程拆解來看，發現當機檢測和初始化的消耗是大頭。要如何改進呢？

從解決思路上來說，包含兩個方面。首先 60 秒的當機檢測，時間太長了。對此，要做到快速發現當機。此外，還要預留資源，當當機出現時，可以省去申請資源，以及初始化的時間。

在當機快速發現方面，我們研發了 Hawk Service，它是一個多數派的連通性檢測服務，具體的檢測流程是 Hawk 叢集中多個工作節點會周期性地檢測叢集中每台機器的連通性，由于它是多數派的，是以可信度是有保障的。最終，Hawk 服務可以做到在 10 秒鐘之内發現一個當機事件。

此外，在預留資源方面，我們擴充了 Flink 作業的資源申請模型，在 Flink 送出時可以設定一個資源備援參數，當備援參數被激活後，會自動保障備援資源量會高于單點故障導緻的資源缺失量，且在資源排布上避免備援資源的聚集性。如圖所示：

有了這兩點能力之後，如果同樣是第三台機器挂掉了，我們能在 10 秒内發現。并且由于資源已經配置設定好了，直接部署一遍作業就可以了。是以整體的恢複時間從 4 個步驟直接縮短為 2 個步驟，時間上從 90s 可以縮短到 15s 左右。

接下來，我們看看如果 Flink 引擎依賴的服務異常了要怎麼辦呢？這裡舉了一個 Kafka 服務異常的例子。還是同樣的 Flink 的作業，依賴兩個 topic，Flink 作業在 B 機房，讀取的 Kafka 也在 B 機房，寫入的 Kafka 在 A 機房。如果出現讀取或者寫入的 Kafka 叢集異常了，Flink 作業需要具備 Failover Kafka 叢集的能力，當然如果是切讀，Kafka 的上遊也需要關聯切流。

在過載場景下，我舉了兩個例子：

• 不同資料源快慢消費導緻滿載

在這個 case 中，消費 topicA 的 source 速度慢，消費 topicB 的資料源快，由于後邊存在 window 操作，會導緻 window 的狀态持續變大，最終引導作業不穩定。這個問題要如何解決呢？

我們采用的辦法是同步所有相關資料源消費的進度，引入一個 source 的協調者（SourceCoordinator），周期性收集 source 源 waterwark 的進展，并根據全局的現狀，預測出來各個 source 源接下來允許讀到的最大位置 target Watermark，之後下發給所有的 source，source 根據得到的 target Watermark 以及目前自己 watermark，确定讀取速度。最終全局 source 達到同步讀的結果，最小 source 和最大 source 的差距在一個可控制的範圍内。

• 硬體資源滿載

如果硬體出現了滿載要怎麼處理呢？例如，其中一個 TM 所在的機器出現 CPU 滿載了，或者大範圍出現機器滿載。

解決方案跟上面的是類似的，控制資料源的消費速度。如圖所示，引入 HealthyCoordinator，周期性檢查 TM 上的資源消耗情況，并根據負載限制 source 的消費速度。動态調節所有資料源的消費速度，進而保證Flink作業的穩定。

3.均衡性

第三個方面，我想跟大家分享一下我們在均衡性上遇到的一個問題。在我們線上叢集的多個機器之間，我們發現最小和最大的機器的 CPU 負載相差至少在 20% 以上。叢集層面的負載不均衡，從穩定上看，可能會觸發作業穩定性下降，從成本上，也會造成資源的浪費。

在解決均衡性問題前，先來看下引發不均衡的因素都有哪些？梳理了下，可能的原因包括：

• Yarn 層面資源排程不均衡
• 作業資源申請不合理，申請過大
• 作業的并發設定不合理或者 Task 排程不均衡，導緻 TaskManager 之間算子 Task 不均
• 資料本身存在不均衡
• 叢集擴容，縮容導緻不均衡

要解決這些問題，我們提了一些改進的方案。

• 改進 Task 排程政策，保障 TaskManager 之間算子的 task 盡可能均衡
• Flink 作業采集實際消耗，重新按照實際消耗向 Yarn 申請資源
• Yarn 保障資源配置設定在機器間均衡
• 在有機器擴容或者縮容時，生産資源消耗均衡的作業調整計劃，進行異步的作業調整

通過以上的政策最終保障 Flink 叢集整體上的均衡性。

五、未來計劃

最後看一下快手在 Flink 上的未來計劃。未來，我們将主要着手于四個方面建設。

• 第一，批流一體模式在更大範圍的推廣應用。例如離線數倉 ETL 的實時化、以及營運活動實時與離線資料的生成。
• 第二，我們會着力推進 Flink 在 AI 資料流上的應用，希望通過 Flink 支撐特征、索引、樣本的實時、離線處理，提效模型疊代的速度。
• 第三，目前有一些線上資料處理鍊路已經使用 Flink 做支撐了，對于 Flink 的穩定的要求也随之上升，我們還需要在穩定性上做持續改進，例如做單點故障的快速 failover 等。
• 最後，由于 Flink 也在支撐線上場景，Flink 需要具備作業内自動且平滑地擴容資源，縮容資源能力。是以彈性伸縮也是我們關注的方向。