一、數據湖架構：從離線數倉到湖倉一體的轉變

數據建設的核心目標一般為：

① 標準統一。

② 可共享。

③ 簡單易用。

④ 高性能。

⑤ 成熟安全可靠。

但是，現在常用來作為實現方案的 Lambda 架構，架構一般如下：

這里存在三個比較嚴重的問題：

① 離線鏈路時效性差。若是直接在這個鏈路上進行提效，則需要的成本比較高。

② 處理邏輯異構。由于目前將實時數據和離線數據分成了兩個鏈路來處理數據，導致很多的處理邏輯無法復用。同時，也會存在一致性的問題。

③ 數據孤島。本身多個鏈路的生產會存在數據孤島，數據無法復用，并且管理相當復雜。

為了解決上述問題，快手使用了數據湖作為數據建設的一個集中式倉儲方案。同時，數據湖也能夠滿足數據建設的核心目標。數據湖具有以下特性：

① 海量存儲。

② 支持可擴展的數據類型。

③ Schema 演進。

④ 支持可擴展的數據源。

⑤ 強大的數據管理能力。

⑥ 高效數據處理。

⑦ 高性能的分析。

業內有很多的數據湖開源實現方案，快手對這些方案的基礎優勢及特性、社區建設情況和技術開發的可擴展程度進行了比較，最終選擇了 Hudi 作為數據湖實現方案。Hudi 在攝入、處理、存儲到查詢，基礎能力支持地比較完善，其具有的多個特點能夠支持數據湖的快速構建和應用，這些特點包括：更新能力強，支持流批讀寫，可插拔的 Payload，支持 MOR 表類型，適配多種查詢引擎，豐富的數據管理操作等。

Hudi 可以幫助快手構建更優的數據鏈路，去完成數據建設的核心目標，架構參考如下：

快手基于 Hudi 構建的數據湖架構具有以下優勢：

① 數據 CURD。優化生產場景模型，提升了整體更新場景的時效；

② 流批讀寫。實現統一的處理，減少了多鏈路多引擎的建設成本；

③ 海量數據管理。對所有的入湖數據進行統一管理，數據平臺的服務和管理方面能夠復用，降低數據的使用成本。

二、基于 Hudi 快速構建快手數據湖：

建設快手數據湖遇到的挑戰以及解決方案

如何使用 Hudi 建設達到核心目標，需要先了解 Hudi 的基本能力：

① 支持不同類型的寫入方式：特別是通過增量寫入和數據合并的兩個基本操作，實現快照的生成；

② 可插拔：可支持所需要的更新邏輯，比如定制化更新模式，可以基于此進行擴展應用場景；

③ 表類型：正如前面提到的，增量寫入和數據合并的操作共同組成快照更新。這兩種基本操作的實現決定了表的類型。不同類型的表，作用不同的應用場景，比如寫多讀少的情況下，選擇使用 MOR 更實時和節約資源；

④ 元數據統計：因為 Hudi 本身實現了更新能力，甚至在之上實現一部分的業務邏輯的，需要保障可描述、可追溯的能力。所以通過元數據的收集和應用，來保證數據的可追溯性；

⑤ 讀取方式：支持 Hadoop 的 inputformat 的方式，兼容常用的查詢引擎，比如spark、trino 等。

使用這些能力，可以為生產鏈路實現提效與統一。

提效主要還是在優化構建離線數倉的時間：

① 比如分層建設時，需要先同步數據，然后再使用離線清洗，再生成后續的數倉的加工數據。現在可以直接一步通過 Flink 任務清洗實時數據，然后使用 Hudi 多級動態分區同步。

② 還有，在離線鏈路生產時，有些數據生產是有更新邏輯的，比如更改部分數據內容。在老的架構下，需要將所有數據都讀取一遍，然后將修改了某幾列的新數據再完全寫入。這里不但時效很差，而且成本消耗很大。現在可以利用 Hudi 的更新能力來高效地更新列數據。

③ 其他的，比如活動的數據需要進行快照分析時，離線鏈路都是小時級別的延遲，一般都需要使用實時鏈路同時生產。使用 Hudi 就可以進行準實時快照的構建，并提供兼容的查詢。

統一的實現，主要是選用了 Flink 引擎作為流批一體的計算引擎，在整體 Hudi 數據湖的生產中進行應用。

通過 Hudi 數據湖架構建設的數據鏈路如下所示：

快手在通過 Hudi 數據湖架構建設新的數據鏈路中，遇到了許多問題。下面，介紹一下快手在建設數據湖過程中遇到的 5 個重要問題以及具體的解決方案。

1、數據攝入的瓶頸

問題描述

快手的數據鏈路都是基于 Flink 生產的，其 Hudi On Flink 架構如下圖所示。

采用上述架構進行數據生產時會遇到性能瓶頸。由于寫入多分區的數據時會通過 BucketAssigner 來進行數據分發，再使用 BucketWriter 實現緩存寫入，那么，當 BucketWriter 之間數據不均衡時，寫入會頻繁觸發溢寫。而當溢寫發生時，又會產生背壓。另外，在提交數據時，由于 BucketWriter 與 Flink 快照進行了綁定，所以 Flink 快照無法實現整點觸發。

解決方案

為了解決上述提到的寫入瓶頸問題，快手優化了寫入邏輯，主要應用于增量數據的同步鏈路。首先，優化寫入模式以提升性能。Flink 寫入方式從緩存寫入修改為流式寫入。寫入數據時不需要緩存，而是直接落盤，并且支持單生產者多消費者的模式，每一個分區文件都可以并行寫入。這樣，可以提高 CPU 的使用率。其次，在攝入的過程中對分發邏輯進行了優化，實現了一個動態感知的模塊。該模塊用于統計數據流量，均衡分發數據到寫入節點，從而保證了各分區之間的數據均衡，來避免某個寫入節點受到過大的數據壓力。

為了實現數據的整點提交，快手實現了自動分區發布功能。根據數據的時間戳生成了分區時間，并且在攝入過程中實時上傳數據的輪轉的時間。在中心協調器里面實現了一個判斷邏輯，如果所有的節點均已完成輪轉，則進行快照的觸發，完成最終的數據提交和分區的同步，來實現整點級的分區發布。

在上述架構中，算子可以進行橫向擴展，整體吞吐量比社區版本提升 10 倍以上，并且能將文件控制在需要的大小（例如：256M）左右。

2、無法使用數據時間進行快照查詢

問題描述

在準實時的數據鏈路上，需要使用 Hudi 的 Time Travel 功能來實現快照查詢。但是，在 SQL 查詢是使用 Timeline 的時間點來進行定位的，而 Timeline 的時間與數據時間不同，且具體的 Timeline 的提交時間在存儲時無法準確感知。

解決方案

快手在 Hudi 的 Time Travel 功能上增加了一個時間版本的元信息。每次寫入時，會通過數據的時間字段來計算數據的版本號。與分區發布過程相同，會實時上傳版本的輪轉時間。在中心協調器判斷是否所有分區已經完成了輪轉，以快照觸發。

由于在提交時存儲快照的數據版本信息，在查詢時，SQL 可以直接使用版本信息來進行查詢。在構建輸入快照的過程中間，會讀取 TimeTravel 的提交信息。這樣，通過判斷數據版本信息是否小于等于 SQL 中指定的時間戳的版本號來構建增量快照，實現某一個時間點的快照查詢。

3、Flink On Hudi 的更新瓶頸

問題描述

在使用 Flink 引擎生產 Hudi 表的過程中，更新是存在一定的瓶頸的。主要體現在，對 Hudi 的不同操作使用的資源是錯配的。比如，寫入操作的寫入內存一般就是攝入的緩存大小。而對于合并操作，合并過程會根據增量數據的數據量來決定 compaction 所需要的內存，一般情況下，這個內存占用量是大于緩存空間的。清理操作中，在構建 FileSystemView 對象時，所占用的內存比較大。

然后，混合操作會影響增量寫入的穩定性。比如合并過程中，并發度無法進行擴展，會導致運行時間長，進而導致快照產生時間延遲（因為快照觸發是需要水位（watermark）下推），甚至會導致任務超時。清理的時候如果遇到異常，也會導致任務的失敗。因此，操作之間的資源復用對操作的執行進度會有影響。

解決方案

解決問題的主要工作是將操作進行分離，支持多種操作并行執行來構建 Hudi 的數據源。

首先，Hudi 支持多種索引。在快手活動期間，會選用 State Index，配置 TTL 來保存一定時間內的快照結果。在需要并發寫入的任務中，由于任務的索引需要相互感知，因此會選用 Bucket Index，可以有效控制寫入緩存資源的占用，而且可以在外部進行操作的運行管理。在表創建時，觸發生成和合并的調度作業；表下線時，自動下線掛載的調度作業。

此外，多個數據源的寫入還需要實現并發控制。首先，對元數據進行加鎖，來避免對元數據的并發操作。然后，在支持并發寫入的過程中，支持了關聯引用，為合并的功能增加了占位邏輯，后續的寫入基于占位合并的 instant，在合并完成之后，基于合并的寫入也是對外可見的，這種方式可以提高寫入的吞吐量。此外，也支持開啟 OCC 控制的并發寫入，在寫入相同的 base 文件時進行并發檢查，支持沖突的失敗回滾或者合并處理，以防止數據結果不正確的現象出現。

4、多任務合并能力不足

問題描述

多任務合并寬表時，在多任務并發運行寫入的場景中，進行索引選擇時，需要考慮到索引數據需要被多任務感知到的因素。若是采用外部索引，則使用成本較高。若是采用 Bucket Index 作為索引，則進行多任務并發寫入時，性能上有優勢。但是，存在一個合并時的瓶頸，這是由于一般情況下，Bucket Index 使用文件大小來控制計算桶數，而合并時使用的資源又取決于增量文件的數據大小，這會導致合并任務的并發度較小，無法滿足合并時的性能需求。在合并的過程中，存在讀取和更新的操作，若是過程中出現了溢寫現象，則整個合并速度會很慢。

在 Schema 的使用方面。不同的寫入任務會使用不同的 Schema，而合并時依賴于寫入任務的 Schema 來生成合并的 Schema 以生成最終的 Base 文件。但是，一些自動上線的寫入任務無法被合并作業感知到。

解決方案

快手實現的并發寫入作業支持了邏輯分桶和多類型合并的能力。邏輯分桶是在物理桶的組織之上進行了二次哈希，本質上是將物理桶分成了更多的桶，在需要寫入時，要先進行桶的排序，并創建對應的索引文件。在后續的合并過程中，基于邏輯桶來生成合并計劃，每一個邏輯桶都會生成一個對應的算子實例。

合并時，作業先讀取物理桶的數據，然后通過索引 seek 到對應邏輯桶的數據位置，之后進行可選擇類型的合并。一般地，在寫入并發已知的情況下，sortMerge 是更快的。在元數據中，增加了合并 Schema 的配置，在寫入時將 Schema 更新到數據源，從而實現了合并Schema的自動擴展和合并任務的自動感知生產。

5、Hudi 生產保障困難

問題描述

Hudi 作為一個較復雜的架構，從生產到運維有比較豐富的支持，比如不同模塊有對應的配置類，支持 metrics 系統、支持 Hudi-Cli 查詢元信息。

但是，這些功能支持在實際生產環境的使用效果并不好。首先，Hudi 的配置過多，使用起來很麻煩。其次，在監控報警和異常中斷的能力上，作為線上服務略顯不足，需要進行自定義強化。因此，快手針對線上需要，加強了生產的保障能力。

解決方案

① 配置精簡。只需要設置一些基本參數（SQL 方式），比如任務類型、保存時間、提交間隔，就可以自動推導生成其他的配置參數。

② 一致性保障。在一致性保障方面，快手自定義實現了 PreCommit 檢驗模塊，比如，會對增量數據的輸入輸出條數進行校驗，同時數據塊的寫入情況在提交之前也會做校驗。

③ 穩定保障。在穩定性方面，快手完善了不同算子的 metrics，包括耗時和數據處理情況，流量吞吐情況等。同時還監控著分區分布，耗時，任務的指標監控來共同保障生產的穩定性。

三、快手的實踐案例

快手數據湖在構建完成之后，在一些具體的業務上進行了應用，取得了明顯的收益。

下面使用四個比較典型的案例，來對比基于數據湖建設的新數據鏈路與舊數據鏈路之間的差異。

最早的時候，核心數倉的 DWD 層生成是需要多層的離線調度任務來進行。在切換到 Hudi 之后，就可以準實時的生成 DWD 層的動態更新數據。此外，還可以根據需要來選擇性的進行數據重分布的操作來對接下來的讀取操作進行提效。這個場景上，快手將離線鏈路升級成了準實時的鏈路，在計算資源上持平，時效上有 50% 以上的提升。

數據在應用過程中，還會有活動數據快照查詢的需求。早期，這些數據若要使用多個數據源進行生產和查詢，需要用到離線鏈路，這種方式的時效性很差，一般會達到小時級。如果想使用實時鏈路加速，需要比較復雜的處理過程。切換到 Hudi 之后，將離線快照的更新時效從小時級縮短到了分鐘級，整體時效達到十多分鐘左右，而且計算資源比以前節省了 15%。在后續的查詢過程中，可以對離線桶的快照數據進行關聯查詢，最終生成需要的活動的結果數據。

生產過程中的數據留存場景中，在生產留存數據時，最早的生產流程是需要用多天的日活數據去重復地生產標簽表，然后與日活的數據 JOIN 生產到最終日活表內，這個過程涉及多次的日活表的讀取和全量數據的回收。切換至 Hudi 后，通過將日活留存狀態直接更新至留存表，數據生產模式從多次的合并生產轉換成了單表生產。在使用當日的日活數據去更新留存表，之前的數據是已經存在的，只需要將日活數據去更新留存狀態即可。這個場景下，鏈路的生產方式上的優化，整體計算資源由于全量讀寫到增量寫入的轉換，根據需求進行定時合并，時效上也有 50% 的提升。

在特征的生產場景內，上游的多個數據生產點合并生產出寬表結果。這個場景下，原來是使用 HBase 來進行合并，在 HBase 中進行行存，在外部用 Hbase 進行生產有一個額外的維護成本，并且需要使用到 HBase 的導入導出工具來進行離線操作，生產鏈路較長。切換至 Hudi 后，可以直接復用已有的生產鏈路邏輯，然后直接對攝入到 Hudi 表內的數據基于并發合并能力構建一張寬表。這個過程中，也可以保證數據是有序的。例如，讀取時可以根據數據需求，比如上游增量寫入數據源的數據已經就緒了，下游就可以直接進行導入。可以理解為實時感知，可以提升整體的處理時效。

可以的觀察結果為，目前的離線分批的合并升級成了準實時單表合并，對時效升級明顯。首先，處理鏈路的計算時間縮短，最長時間節省 5 個小時；在鏈路計算過程中所占用的臨時存儲空間和計算資源得到了節省，同時，也節省了 HBase 集群所需要的開銷。

四、快手的發展規劃

快手數據湖當前還有一些待優化的工作。首先，缺少完善的元數據和數據管理服務。在查詢模式上，由于不支持實時表，也還沒有達到離線和實時查詢的統一。此外，當前快手數據湖生產方式還沒有做到無感知的兼容，所以主要在新的場景上使用，總體的使用率占比不高。

未來快手數據湖將作為統一存儲的技術組件，支持更多類型的數據以及拓寬數據湖支持的表類型，例如實現類似于實時表的定義。完善數據的管理來提升數據組織的合理性。實現兼容已有鏈路的輕量切換方案。將實現流批一體的數據生產，提供高效統一的查詢能力作為數據湖建設的最終愿景。

五、問答環節

Q1：詳細介紹一下加鎖的部分。

A1：其實社區本身也支持 OCC 機制，其實現邏輯是，在寫入過程中對元數據進行加鎖，在最終的提交階段，會對寫入文件做 CAS 操作，通過對比來發現沖突。若是發現兩個寫入任務寫入同一個 base 文件的情況，即表示寫入任務之間存在沖突，會將后寫入的作業標記為失敗。快手也使用了這個機制來避免并發寫入時，寫入同一個 base 文件，影響最終結果。

快手在這個基礎之上，對合并和更新過程做了一個優化。比如說，Flink On Hudi 架構的準實時寫入過程中，若按照社區的寫入邏輯，將合并和更新識別為兩種操作，會導致合并阻塞了整個寫入操作，或者合并操作一直無法完成，會導致讀取數據源的效率低。經過快手的優化后，寫入過程中，寫入效率不會受到 Compaction 結果的影響。因為，快手會使用之前合并執行計劃的基于時間戳的占位符來進行寫入操作。社區的默認邏輯是基于 baseTime 來生成數據，這使得合并的結果和寫入數據之間可能存在沖突。若是采用占位的合并計劃的 Instance 與已有的數據不會產生沖突。但是需要保證合并操作必須完成，否則后面的寫入數據是不可見的，通過這種方式，可以提升整體的增量寫入的吞吐量。

Q2：Compaction 資源錯配的問題怎么解決？異步的 Compaction 是否有相關的經驗可以分享一下？

A2：資源錯配的主要原因是寫入操作需要的資源和并發資源不一致。若是將寫入過程各個操作分離開，那么可以根據寫入任務的流量情況來調整寫入資源。快手目前仍然采用默認的配置，寫入操作的一個 TaskManager 占用了 6G-8G 的內存，其中包含了多個并發寫入的 Slot。在合并過程中，取決于需要的時效。比如說，需要整體的合并時效比較高的話，需要盡量的避免溢寫現象的發生，這時，需要將 Compact 的內存設置得比較大。默認情況下，快手將合并任務調整為 4 核 10G 左右。當有寫入數據量比較大且合并速度快的需求時，則需要將內存設置更大一點，這樣增量數據基本上存儲在合并操作的內存中。這樣，1 到 2G 的文件可以在 10 分鐘以內處理完成。