作者 | 張靜

編輯 | 姚佳新

本文整理自快手?jǐn)?shù)據(jù)架構(gòu)研發(fā)專家張靜在WOT2023大會上的主題分享，更多精彩內(nèi)容及現(xiàn)場PPT，請關(guān)注51CTO技術(shù)棧公眾號，發(fā)消息【W(wǎng)OT2023PPT】即可直接領(lǐng)取。

今天的分享分為四個部分：首先介紹傳統(tǒng)離線鏈路，它存在哪些痛點(diǎn)；第二部分引入數(shù)據(jù)湖的特性；第三部分是通過快手?jǐn)?shù)據(jù)湖幾個典型的業(yè)務(wù)場景來說明如何基于數(shù)據(jù)湖技術(shù)重塑離線鏈路的生產(chǎn)；最后一部分介紹近期工作和長遠(yuǎn)規(guī)劃。希望通過本次分享能夠讓大家了解數(shù)據(jù)湖技術(shù)在重塑離線生產(chǎn)方式中的關(guān)鍵作用。

一、傳統(tǒng)離線鏈路的缺點(diǎn)

快手的傳統(tǒng)離線鏈路和很多公司是一致的，基于 Hive做離線分層數(shù)倉的建設(shè)。在入倉環(huán)節(jié)和層與層之間是基于 Spark 或者 Hive做清洗加工和計(jì)算。這個鏈路有以下四個痛點(diǎn)：

更新成本高：Hive 表最細(xì)的更新粒度是分區(qū)級，需要先掃出分區(qū)的全量數(shù)據(jù)，關(guān)聯(lián)這次更新的增量數(shù)據(jù)得到這次的全量數(shù)據(jù)并覆蓋原來的分區(qū)。這個過程導(dǎo)致計(jì)算開銷比較大，且降低時效性；

缺少索引：不僅影響更新，也影響讀取。因?yàn)椴樵兇蟛糠謺話呙铻橹鳎纱藭?dǎo)致查詢效率低；

缺少事務(wù)：多個寫入任務(wù)之間，寫入任務(wù)和讀取任務(wù)之間缺少事務(wù)機(jī)制，需要讀寫鎖來避免數(shù)據(jù)的不一致；

啟動調(diào)度晚：目前離線任務(wù)調(diào)度最細(xì)粒度是小時級別，會影響下游各層的數(shù)據(jù)可見性；

圖片

二、HUDI 數(shù)據(jù)湖的特性

針對傳統(tǒng)離線鏈路的缺點(diǎn)，我們決定引入數(shù)據(jù)湖來解決上述的痛點(diǎn)。快手是在21年開始探索數(shù)據(jù)湖方向，我們進(jìn)行了技術(shù)選型，考慮到HUDI 對更新能力的支持，以及活躍的社區(qū)生態(tài)，由此便選擇了HUDI。HUDI 具備如下幾個特點(diǎn)：

寫入：由于 HUDI提供多種內(nèi)置的索引，基于這些索引可以提供高效的更新能力；寫入支持流式入湖，也支持離線入湖；支持多種的寫入操作，比如插入、更新、刪除、覆蓋；支持多種輸入源，比如更新流，日志流。

查詢：支持多種的查詢方式，比如讀優(yōu)化查詢、快照查詢和增量查詢；提供時間旅行的特點(diǎn)解鎖查詢歷史版本的能力；社區(qū)做了很多優(yōu)化提高查詢的效率。

并發(fā)控制：HUDI 引入 MVCC 來控制寫入任務(wù)和查詢?nèi)蝿?wù)之間的并發(fā)。引入 OCC 來控制多個寫入任務(wù)之間的并發(fā)。同時社區(qū)也有一些關(guān)于無鎖的并發(fā)控制。

豐富的表服務(wù)：Compaction、Clustering、Clean 等。

開放性：適配多種計(jì)算引擎和查詢引擎，比如 Spark，F(xiàn)link，Presto，Trino，Starrokcs，Doris 等

Schema Evolution：提供Schema 演進(jìn)的能力。

圖片

三、快手?jǐn)?shù)據(jù)湖的典型業(yè)務(wù)場景

下面通過快手在數(shù)據(jù)湖上的幾個典型業(yè)務(wù)場景介紹如何用 HUDI重塑離線鏈路產(chǎn)生。分為三個方向：數(shù)據(jù)同步、數(shù)據(jù)更新、寬表拼接。每個方向都會介紹兩類最有代表性的場景。

圖片

1.數(shù)據(jù)同步 – 日志流入湖

首先是數(shù)據(jù)同步里日志流入湖。快手內(nèi)部的數(shù)據(jù)同步工具有一個限制：只支持日期和小時兩級分區(qū)。所以一個日志流從 Kafka 到入倉整個鏈路需要多個離線任務(wù)加工，這就導(dǎo)致了鏈路長，重復(fù)計(jì)算和冗余存儲的問題。

圖片

基于 HUDI 改進(jìn)后的方案，整個鏈路得到極大的簡化。直接用 Flink 任務(wù)做日志流數(shù)據(jù)入湖。最后一層將 HUDI 表落到 DWD 層數(shù)據(jù)主要是做兼容性，這樣下游業(yè)務(wù)依然可以訪問原來的 Hive 表，同時獲得時效性的提升，在資源持平情況下，時效性從之前1h40min縮減到40min，也降低了了鏈路的復(fù)雜度。

圖片

2.數(shù)據(jù)同步 – CDC 數(shù)據(jù)入湖

第二個場景是更新場景入湖。歷史上 Mysql to Hive的方案有兩個鏈路，一個全量初始化任務(wù)，一個是增量同步任務(wù)。初始化任務(wù)把全量數(shù)據(jù)落到一個HIVE 全量快照表，完成后啟動增量同步任務(wù)把增量binlog 數(shù)據(jù)落到一個 HIVE增量表，每天合并前一天的全量和今天的增量生成一個新的全量快照表。

圖片

Mysql to Hive 方案的痛點(diǎn)是時效低。時效低有兩方面原因：第一個是離線任務(wù)調(diào)度周期是T+1級別，第二個是任務(wù)調(diào)度以后才做全量和增量的合并。

圖片

改造后的Mysql to HUDI，鏈路得到了簡化，直接把 CDC 更新數(shù)據(jù)落到一個 HUDI 表里，這個 HUDI 表是沒有日期或者小時分區(qū)的。內(nèi)部的 MySQL to HUDI 和其他公司的 CDC 更新流入湖比較起來有一些差異化的需求，因此我們在設(shè)計(jì)上也是有所不同。

避免在全量同步完成后再啟動增量同步任務(wù)：因?yàn)椴捎脗鹘y(tǒng)的串行調(diào)度，如果全量同步任務(wù)執(zhí)行很久才結(jié)束，增量同步啟動后可能發(fā)現(xiàn)最開始的一些 Kafka 數(shù)據(jù)已經(jīng)被清理了，導(dǎo)致數(shù)據(jù)丟失。因此，支持全量初始化任務(wù)和增量同步任務(wù)的并行，不需要等全量初始化任務(wù)完成后再去調(diào)度增量同步任務(wù)。

按照事件時間來查詢某個版本：HUDI 的版本是一個 processing time 的語義，但是用戶需要能按照 event time 語義來訪問某個 HUDI 版本。為了支持按照事件時間方案，在元數(shù)據(jù)里維護(hù) Processing time 到 Event time 的映射關(guān)系。收到按照事件時間的快照查詢請求，先做一下映射得到 processing time，再基于time travel能力查詢對應(yīng)的版本。

數(shù)據(jù)就緒后盡快發(fā)布對應(yīng)版本：如果完全依賴周期性的 checkpoint 來做分區(qū)發(fā)布會導(dǎo)致數(shù)據(jù)就緒后不能立刻發(fā)布對應(yīng)的版本。這里修改了 Flink 引擎的邏輯，除了周期性的 checkpoint 以外，又增加一種非周期性的checkpoint 用于監(jiān)聽到整點(diǎn)數(shù)據(jù)就緒以后立刻發(fā)布分區(qū)。

兼容當(dāng)前 HIVE 表的使用方式：1. Mysql to HUDI 鏈路里的HUDI 表是沒有日期分區(qū)，如何能按照日期分區(qū)查詢。2.長生命周期管理，用戶可能需要訪問很久以前的數(shù)據(jù)。為了支持這兩個需求，Mysql to HUDI 的鏈路會輸出兩個表，一個是無時間分區(qū)的 HUDI 表，一個是HIVE 表。在發(fā)布分區(qū)時，會在HIVE 表里添加一個新分區(qū)，這個時候分區(qū) location下是沒有數(shù)據(jù)，分區(qū)元數(shù)據(jù)里維護(hù)了它對應(yīng)哪個 HUDI 表的哪個版本。無時間分區(qū)的HUDI 表是沒有辦法直接做長生命周期的，所以定期把HUDI 數(shù)據(jù)同步到Hive 表中去。歸檔后的 HIVE 表分區(qū)就是一個普通的 HIVE 分區(qū)，它的 location 下有對應(yīng)的分區(qū)數(shù)據(jù)。因此，這個HIVE 表是一個異構(gòu)的HIVE 表。異構(gòu)性體現(xiàn)在兩個方面，第一個元數(shù)據(jù)是異構(gòu)的，第二個是數(shù)據(jù)是異構(gòu)的。這個異構(gòu)設(shè)計(jì)對用戶是透明的。當(dāng)用戶查詢HIVE分區(qū)的時候，引擎通過 Hive 元數(shù)據(jù)判斷這個日期是否被歸檔，如果還沒有被歸檔，會通過分區(qū)元數(shù)據(jù)里的HUDI 表和版本把請求路有到HUDI 表上。如果是歸檔后的分區(qū)，直接走正常的HIVE查詢流程把分區(qū)數(shù)據(jù)返回給用戶。

圖片

Mysql to HUDI的整個鏈路如上圖。分為左右兩部分。左邊是必選的，做CDC 入湖；右邊是可選的，為了支持兼容HIVE 的需求。

3.數(shù)據(jù)更新

數(shù)據(jù)更新的第一個業(yè)務(wù)場景是人群包圈選。每次活動DAU 是一個非常重要的指標(biāo)，人群圈選業(yè)務(wù)是根據(jù)用戶的歷史行為來圈選出一些潛在的目標(biāo)用戶。歷史方案是基于天級離線數(shù)據(jù)和小時級離線數(shù)據(jù)組合計(jì)算生成。這種方式存在的最大痛點(diǎn)就是時效性問題，某些場景下的小時級產(chǎn)出的數(shù)據(jù)延遲在3-4 小時左右，對于除夕活動來說，這種延遲是不能忍受的。基于 HUDI 改造后的鏈路是用一個實(shí)時的 Flink 任務(wù)，在入湖過程中完成更新。這使得整條鏈得到簡化，不僅時效性從3h ~ 4h左右縮短到15min左右，而且資源也有節(jié)約。

第二個業(yè)務(wù)場景是基于HUDI 自定義的payload能力的N天留存標(biāo)簽更新。歷史的留存鏈路加工流程需要大規(guī)模Join 并且需要與行為數(shù)據(jù)進(jìn)行整合，并且需要大規(guī)模數(shù)據(jù)回刷。具體過程是用當(dāng)天的日活數(shù)據(jù)和歷史N天的日活數(shù)據(jù)算出當(dāng)天日活用戶在過去 180 天的留存標(biāo)簽，存一個中間表。然后分別用過去N天的行為數(shù)據(jù)關(guān)聯(lián)這個中間表得到最新的標(biāo)簽覆蓋回對應(yīng)的分區(qū)。這個方案的缺點(diǎn)是時效低，重復(fù)計(jì)算和重復(fù)存儲。

基于HUDI 改造后的鏈路從剛才的多層關(guān)聯(lián)升級為單表生產(chǎn)，時效性也是有了很大的提升，從2.5h縮短到1.5h。資源開銷也是有收益的。這里最重要的就是基于 HUDI 的 MOR 表能力和自定義payload 的特點(diǎn)。寫入流程非常輕量，將當(dāng)天的日活數(shù)據(jù)產(chǎn)生的增量數(shù)據(jù)寫到歷史N 天的分區(qū)里。合并流程做在分區(qū)內(nèi)部做局部關(guān)聯(lián)只更新對應(yīng)的留存標(biāo)簽。

圖片

圖片

4.寬表拼接

第三個方向是寬表拼接，也介紹兩個典型的業(yè)務(wù)場景，一個是離線寬表模型，一個是準(zhǔn)實(shí)時的多流拼接。

寬表模型是指把業(yè)務(wù)主題相關(guān)的指標(biāo)、維度、屬性關(guān)聯(lián)在一起的一張大寬表。寬表模型因?yàn)榻Y(jié)構(gòu)簡單，模型可復(fù)用度高，數(shù)據(jù)訪問效率等優(yōu)勢，廣泛地使用在 BI 和 AI 場景。

圖片

基于 HUDI 的寬表拼接之前有很多公司也有分享，我們內(nèi)部的寬表拼接有一些差異化的需求。

支持多個寫入任務(wù)并行：允許多個寫入任務(wù)并行加工一張寬表，每個寫入任務(wù)加工這個寬表中的部分列。

支持 Schema Evolution：在業(yè)務(wù)演進(jìn)過程中可能隨時需要有更多的列加進(jìn)來。用戶希望在創(chuàng)建表的時候，只需要定義必要的列，比如主鍵列、分區(qū)列、排序列。后續(xù)可以很靈活地添加新的列。

支持 Implicit Schema Evolution：顯式的 Schema Evolution 是指通過類似于 Alter table add column 這種DDL 語句來修改表。Implicit Schema Evolution，是指在寫入任務(wù)的 Schema里包含了表里不存在的列，會在寫入任務(wù)提交時追加到這個表的最后。

支持 Partial Insert：寫入任務(wù)不需要指定表里的所有列，允許只插入表里的部分。

支持不同分區(qū)設(shè)置不同的桶個數(shù)：有一些業(yè)務(wù)分區(qū)存在非常大的數(shù)據(jù)量差異，所以需要能支持不同子分區(qū)設(shè)置不同的桶個數(shù)。

支持快照隔離：讀取任務(wù)和寫入任務(wù)之間支持快照隔離，上游加工好部分列以后，下游就可以先讀這些加工好的部分列。

圖片

上圖是一個簡單的寬表拼接的例子。兩個寫入任務(wù)加工一個寬表，第一個寫入任務(wù)加工 id, ts 和name。第二個寫入任務(wù)加工 id, ts 和 price。每個寫入任務(wù)只需要寫入部分列，這個是 partial insert 的能力。最后合并流程做拼接。另外，這個圖也可以說明 schema evolution。建表時，只定義了主鍵、排序鍵和分區(qū)鍵。第一個寫入任務(wù)提交的時候追加了name 列，第二個寫入任務(wù)提交的時候追加  price 列。

圖片

寫入階段分為兩個階段，第一個階段寫入數(shù)據(jù)，第二個階段提交數(shù)據(jù)。第一個階段是無鎖方案的設(shè)計(jì)，第二個階段是有鎖的設(shè)計(jì)。第一個階段，寫入任務(wù)是在加工同一個文件組的同一個數(shù)據(jù)版本下不同的增量文件來避免多個任務(wù)把一個文件寫花。在提交階段引入一種特殊的沖突檢查機(jī)制，允許在不同分區(qū)或者是相同分區(qū)的不同列上的并發(fā)寫入，另外這個階段按需更新 schema，發(fā)現(xiàn)有新增的列需要更新schema 。

這個方案也可以用在實(shí)時寬表拼接場景，這里因?yàn)闀r間關(guān)系，不再做贅述。最后說一下在目前的寬表拼接實(shí)現(xiàn)里有一個限制，即寫入任務(wù)正在進(jìn)行時不可以生成合并計(jì)劃，可能存在丟數(shù)據(jù)的風(fēng)險。在用戶角度這個限制有三點(diǎn)影響：第一個是離線寬表拼接場景需要依賴任務(wù)以來關(guān)系來避免寫入任務(wù)和 schedule compaction 的并行。第二個是對實(shí)時寬表拼接場景，只能在同一個 Flink 作業(yè)的多個 pipeline 里共同加工一個寬表，不能多個 Flink 作業(yè)同時加工一個寬表。第三個是不能滿足實(shí)時和離線任務(wù)共同加工一張寬表的需求。

四、未來規(guī)劃

圖片

近期的工作有四點(diǎn)：

（1）Schedule Compaction 和 Writer 的并發(fā)。

（2）可擴(kuò)展的 Bucket index，實(shí)現(xiàn)根據(jù)數(shù)據(jù)量自動適配 bucket number 個數(shù)。

（3）加速寫入流程：這里涉及到多個優(yōu)化點(diǎn)，一個是優(yōu)化寫入鏈路，一個是減少序列化和反序列化開銷

（4）服務(wù)化建設(shè)。包括 MetaStore Service 和 Table Service。

圖片

中長期的工作圍繞兩個方向，第一個是建設(shè)實(shí)時數(shù)據(jù)湖。對于實(shí)時數(shù)據(jù)湖也會有很多挑戰(zhàn)，需要把它補(bǔ)充齊才可以把實(shí)時化做起來，這塊會引入流計(jì)算領(lǐng)域領(lǐng)域通用的概念，比如事件時間和watermark。第二個是基于HUDI的分析查詢場景。我們會參與到社區(qū)的建設(shè)中，通過構(gòu)建物化視圖減少重復(fù)計(jì)算加速查詢，后續(xù)也會引入緩存加速分析查詢的場景。這兩個方向都有很多地方需要探索和完善。