一、背景與挑戰

首先來介紹一下相關背景。

傳統數倉存在實時和離線兩條鏈路，來滿足業務對于時效數據的時效性和數據量的不同需求。離線會維護歷史的全量視圖，實時會維護增量視圖，最后在服務層去進行數據的匯總，從而支持后續的在線的serving、 OLAP 查詢以及看板的應用等等。 

因為處理場景的差異，在實時和離線數倉的具體實現上，依賴的底層存儲計算引擎基本上是完全隔離的，實時依賴的主要是以 Flink 為代表的流式計算引擎來做計算，而離線依賴主要是以 Spark 為代表的引擎，實時主要依賴 KV 或 MQ 這樣的多種存儲選型。離線則常常采用 Hive 為代表的存儲引擎，傳統的數倉架構，它本質上結合了流計算和 批計算的優勢，通過兩套代碼來兼容實時數據和離線數據的優勢，彌補各自的缺點。但是這兩套架構或代碼也帶來了兩倍的資源成本，并且因為底層計算引擎的不同，對于相同算子的處理的語義不是完全一致的，它們的計算結果就會存在差異。所以對于研發同學來說，這部分差異也會給數據校驗等其它一些工作帶來額外的負擔。總結下來，傳統數倉的 Lambda 架構主要存在三個問題要解決：第一個是一套計算邏輯，但需要寫離線和實時兩套代碼，帶來了兩倍的運維成本；第二是離線實時兩條鏈路帶來了資源冗余的問題，雙倍的資源的成本；第三是兩套引擎計算口徑不完全對齊，導致數據校準方面會有比較大的困難。所以業內提出了希望通過批流一體來解決當前傳統 Lambda 架構的問題。基于上述問題，再結合內部的實際場景，批流一體的訴求可以分為兩種，第一種是計算的批流一體，第二種是存儲的批流一體。計算的批流一體指的是希望通過底層一套系統，業務層的一套代碼同時滿足離線和實時的開發需求，從而解決兩套系統帶來的研發效率、人工成本、運維成本等資源成本相關的問題。另外我們希望一套系統能夠對齊計算指標的口徑來解決數據一致性的問題。

存儲的流批一體包括，第一是實時和離線所在的存儲統一，第二是實時和離線的數據能夠復用。實時和離線存儲統一是指我們希望實時和離線能夠使用統一選型的存儲，這就要求存儲能夠滿足大規模、全量和增量數據的讀寫訴求。批流數據復用指的是批處理能夠使用流處理的結果數據，提升整個離線數倉的產出時間。典型的 case 就是 ODS 層的數據復用，另外流處理也能夠復用批處理的數據來解決鏈路冷啟動的時候，需要將離線的數據回灌到實時存儲中的額外成本，我們通過 LAS 去實現了批流一體的能力。在計算層，我們的解決思路是對外暴露統一的 SQL，底層根據 SQL 處理的場景選擇不同的執行引擎去執行，對用戶屏蔽以底層執行引擎的差異性。之所以這樣做的原因是我們認為不同引擎適用于不同場景，很難找到引擎能夠同時滿足實時和離線，不同時效性和數據規模的要求，在 SQL 層，我們對齊了底層執行引擎的差異性來實現計算口徑的對齊，解決了上面說到的一致性問題。在存儲層，我們基于湖倉一體的架構，通過數據湖實現了批流一體存儲的能力。除了能夠支持流式的增量和批式的全量讀寫之外，我們還支持了高效的 OLAP 查詢能力以及維表 join 的能力。

既然提到了LAS，我們看一下 LAS 的整體結構， LAS全稱是 Lakehouse Analysis Service。湖倉一體的數據分析服務，融合了湖與倉的優勢，既能夠利用湖的優勢將所有的數據都存儲到廉價存儲中，供積極學習、數據分析等場景使用，又能夠基于數據湖構建數倉，讓BI、報表等業務場景去使用。 

LAS具有如下一些特性：首先是能夠支持統一的元數據，避免在數據湖中存在數據孤島的問題，每個數據都是可追溯的。第二個是依托數據湖提供ACID 的能力。第三點是機器支持企業級的權限管控。第四點是支持資源的極致彈性擴縮容，降低用戶的使用成本。最后是引擎的內核的極致優化，提供高效的讀寫性能。LAS 整體架構，最上面一層是湖倉開發工具，為數據應用場景提供能力支撐。下面一層是數據分析引擎，支持流批一體SQL，解決計算批流一體的問題，并且支持根據 SQL 的特點去做引擎的智能選擇和加速。針對 OLAP 分析，我們會將 SQL 路由到不同的執行引擎去執行，比如對 Ad-hoc我們會用 Presto 去進行查詢。再往下一層是統一元數據層，最后一層是基于 Hudi 去實現的流批存儲層。本文會聚焦在流批一體存儲的細節實現上。

我們需要分析現有的離線數倉和實時數倉的具體需求，來考慮流批一體存儲的實現方式。離線數倉的整體結構分層相對來說還是比較清晰的，使用的存儲也會比較單一，主要是Spark 加 Hive 的形式，提供高效的數據處理和吞吐能力，能夠支持離線數據回溯場景下的并發更新。但是實時數倉的使用存儲相對來說會復雜一些，一般會依托 Kafka 或 MQ 進行每一層數據表的構建，為了支持高效的 join 的性能，在維表的存儲選型上，我們往往會根據數據量的差異選擇KV、Hbase 或者是 MySQL 去進行存儲。在整體的 DWS 數倉鏈路梳理完之后，到了數據應用層會對接 ClickHouse、Doris這樣的高效的 OLAP 引擎，去對外提供計時的數據看板報表等等。數據應用層還會有一些 serving 的功能，服務層會將數據寫到 KV 或者 MySQL 或者 ES 這些存儲里，對外提供 serving 的服務。

在構建實時數倉新鏈路的時候，對于鏈路冷啟動，需要使用歷史分區的數據，所以我們需要將離線的數據回灌到實時鏈路的 MQ 里面，受限于 MQ 帶寬的限制，整體的回溯周期可能會非常的長，并且操作很復雜。另外當計算指標有問題，或者是整體的計算口徑需要調整的時候，也需要使用離線的數據去對實時數據進行回刷，同樣它也會遇到回溯周期長，操作復雜等一系列問題。通過介紹可以看出實時數據倉庫整體相對比較復雜，存儲方式和構建標準沒有完全統一。為了更精細地分析實時數據倉庫對于流批一體存儲的需求，我們基于數據量延遲、數據一致性要求和計算周期等維度，將場景劃分成了三類：日志計算、長周期計算和全量計算。

日志計算的場景特點在于數據量比較大，但是可以接受少量數據丟失。大部分數據要求在分鐘內計算，并進行分組聚合。但該場景的痛點在于希望通過批流數據復用和統一以提升數據時效性和降低資源成本。對于長周期計算場景，數據量相對中等。需要對指標進行復合計算，但整體數據周期可能較長。直播類業務場景可能持續一個月，數據要求在秒級。該場景的痛點在于冷啟動和回溯過程復雜、周期長、成本高。全量計算場景數據量不大，會將全量數據存儲到Flink state中進行計算，要求強一致性，時效性要求在秒級別。但該場景遇到的最大問題在于因數據存儲到Flink state中，未進行分層結構，回溯的中間結果可能不透出，不太利于開發人員進行調試操作。總結來看，數倉對于存儲的主要需求可以概括為以下幾點：其一，實時存儲不統一，運維復雜；其二，實時離線存儲不統一，資源成本高；其三，冷啟動或回溯過程復雜耗時；其四，無法查詢中間數據。因此，我們的批流一體存儲方案，不僅要解決上述痛點，還需要具備以下基本能力：支持離線回溯場景的分區并發更新，且數據讀寫吞吐量不低于Hive。對于流式場景的流批處理，需要滿足低延遲的要求，數據延遲約為幾秒鐘，并能夠提供高吞吐量以支持千萬級RPS。此外，我們還需要提供支持Exactly-once和At-least-once數據一致性語義的功能。為了實現整體的流批一體的目標，還需要支持多引擎，例如Spark、Flink的讀寫，同時也需要支持多種OLAP引擎進行查詢。

二、設計方案

接下來看一下我們的流批一體存儲方案，結合剛剛討論的流批一體存儲目標，我們發現現有的基于數據湖倉一體的架構，實際上已經可以滿足大多數要求了。當前數據湖倉一體的架構已經支持所有數據入湖，并支持Spark、Flink引擎，同時也可以進行離線和實時的數據操作。在下游數據應用方面，數據湖倉一體的架構還支持ihook、metastore、adhoc等OLAP查詢方式。在字節內部使用的場景中，業務會通過Flink實時將數據入湖，使用Spark批量回溯更新湖內的數據，并且下游會使用Presto查詢服務來觸及下游的看板。因此，我們信息湖倉庫主要使用了Hudi這樣的開源方案。在功能方面，數據湖倉庫基本符合了實時和離線數倉對于流批一體存儲的需求，而這主要是因為Hudi本身提供了事務支持，我們在內部還進行了桶索引機制的優化以進一步提高入湖的性能，并且通過metastore的元數據服務來支持并發寫入功能。此外，Hudi原生支持多引擎，因此既可以對批流進行讀寫消費，也可以使用Presto進行交互式分析。

在內部，湖倉一體架構大規模地落地了離線的數倉場景和部分近實時的數倉場景。但是因為 Hudi 本身的或者數據庫本身分鐘級別的可見性，它還是沒有辦法做到實時數倉存儲的標準方案。

為了解決時效性的問題，提供低延遲的能力，我們內部自研了基于內存的服務，它構建于數據湖之上，形成了一套整體的高吞吐、高并發、低延遲的實時數據服務方案。底層方案的整體架構如圖所示。底層是持久化數據層，會復用Hudi當前的能力持久化數據，文件分布跟 Hudi一致，通過 log 的行存文件和 base 的列存文件進行數據存儲，會通過 file slice 這種基于時間戳的方式去維護數據的版本信息，通過 file group 這樣的方式去對文件進行分組，相同組件的數據會存儲在同文件組內。這種文件變分組的方式，再結合索引的能力，能夠有效地提升數據入湖的性能和查詢的性能。

上層服務層主要分為兩個組件， BTS 和 Table Service Management。BTS 是基于內存構建的服務層，它主要是來解決實時場景下數據讀寫的時效性的問題，通過內存去對數據讀寫進行加速，TSM (Table Service Management)，是表優化的服務，它會異步地去執行一些表優化的操作，從而實現對查詢的加速。

這里的表優化操作指的包括社區原生的壓縮聚合 clustering，以及一些索引異步構建，視圖異步構建的一些操作。壓縮聚合指的是對日志文件和基礎文件進一步合并以生成新的列式存儲文件，這樣對整體查詢效率而言更優。而 clustering 則是合并小文件以減少文件開銷。當前 TSM 只支持這兩種能力以及清理能力，我們計劃結合社區現有的 MDT 能力來異步構建多級索引，以提升交互式查詢的性能。

表優化操作是一個完全異步的過程。這部分是我們自主開發的服務，因為一些社區原生實現并沒有做到完全異步。為什么要異步呢？因為 compaction 和 clustering 的執行時間比較長，同步操作會影響數據湖的寫入速度，特別是在實時場景下不可接受。而社區的異步操作僅指寫入時不阻塞，但是 compaction 會共享寫入資源在同一個應用程序內執行。這可能會影響寫入作業的穩定性，因此我們在內部落地過程中發現了這個問題，最后實現了一個完全異步的調度執行，同時不共享寫入資源的服務。在具體的執行層面上，我們還利用混部的資源以降低成本。

1、數據組織形式

基于這樣的新的流批存儲架構，我們新增了中間的服務層，特別是BTS這樣的實時元數據加速層，整體的數據組織形式如下圖所示。數據組織形式在邏輯層分為表分區、文件組和文件大小等概念。數據寫入時，先寫入對應分區，再根據主鍵寫入對應文件組。文件組的底層文件存儲分為內存數據和分布式文件系統數據兩種類型。在內存中，數據由塊構成，而在分布式文件系統中，數據的組織形式與Hudi相似，采用基礎文件和日志文件的模式。值得注意的是，我們引入了日志存儲層來存儲WAL文件，以確保數據在寫入過程中的有效性，并解決內存數據丟失的問題。因此，在寫入內存數據之前，我們會先寫入WAL文件，確保這部分數據已被持久化存儲，實際操作才被認為是成功的。對于內存文件塊與持久化存儲文件之間的映射關系，每個塊對應一個WAL文件以保證數據的容錯性。每個內存中的塊都不會永久存儲在內存中，而是定期地刷到持久化存儲文件中。在實際操作中，多個塊通常對應一個日志文件。

2、數據讀寫方式

再來看一下整個數據讀寫的交互方式。首先，做了流批復雜的分離，因為流場景和批場景對于數據的可見性和時效性的要求是不完全一致的。對于批量回溯的場景，用戶并不是希望能夠馬上可見，只是希望把這部分數據做好更新和校準而已。

在批量數據更新的場景中，數據會直接寫入持久性存儲中，也就是會寫入到HDFS上，而不是通過內存。這種方法可以極大地提高我們的讀寫吞吐量。對于流式讀寫的情況，會首先訪問BTS之類的內存服務進行讀寫。這里的主要實施細節是，在寫入數據時，我們會優先寫入內存中，會首先寫WAL文件以保障容錯。在讀取數據時，由于內存中的數據不會一直存在，因為cache會定期清理，所以讀取時會優先訪問內存中的數據。如果發現內存中沒有數據，我們會先加載WAL文件并對其進行預加載，以盡可能地將數據優先加載到內存中，以保證流式讀取的時效性。如果發現WAL文件也不存在，或者被清理了，那么我們就會轉而去讀取持久性存儲中的日志。在大多數情況下，這是流式讀寫，整個周期相對較短。因此，在內存中，WAL文件的存儲能夠做到一周之內的時效性。一周之內的用戶都可以正常從內存中消費這部分數據。如果用戶希望存儲長周期的數據，那么他可能需要承擔更多的存儲成本。我們需要盡可能避免從HDFS上加載日志文件，這是整體的數據讀寫方式。

3、BTS架構

剛剛提到的流讀的場景，我們也做了讀寫的負載分離，會有單獨的讀集群去承接整體的讀流量，來避免它影響寫入節點的性能。我們來看一下整體的BTS架構，BTS首先是Master-Slave架構。Master主要負責一些元數據信息的管理，它的結構與HDFS相似。Table Server以Slave的形式存在，負責數據的讀寫，并在其上存儲由若干個block組織而成的文件。對于Master，它管理的元數據包括Table Server的信息以及block的元數據管理。由于元數據管理的需要，它必然會引入一定的負載均衡機制。因此，我們目前實現了比較簡單的負載均衡機制，旨在避免某一Table Server內存被打爆的情況。

Table Server主要提供數據讀寫能力，維護本地的塊并定期進行塊清理。它異步將某些塊刷新到HDFS上，整個數據讀寫流程是客戶端請求master，獲取需要寫入的塊，然后找到對應的Table Server進行數據寫入。在寫入時，它優先寫WAL文件，再寫內存文件。當數據全部寫入并ACK返回后，表示這批數據已經成功寫入，不會再丟失。此時涉及到數據提交問題。這部分統一由主節點master去管理事務，其整體的事務機制與Hudi目前的實現相似，即依托于引擎進行提交。對于Flink來說，每次checkpoint都會觸發主節點進行提交。因此，當下游消費這批數據時，如果我們需要達到秒級數據，則不太可能進行秒級信息源數據的提交。因此，在這部分下游可能會讀取一些基于read on committed的數據，所以需要進行去重操作，以確保At least once的語義。以上就是整體的BTS結構的介紹。接下來介紹落地場景。

三、落地場景

我們的主要落地場景是流式數據計算，它類似于離線數倉，需要進行一些ETL清理和簡單的聚合計算。左側是整體架構方案，我們使用了基于Hudi加BTS的數據湖方案來替換MQ，從而實現每一層數據表的存儲。在維表上，我們目前仍使用KV存儲。我們當前的目標是替換MQ場景。

原先離線需要將實時表從 MQ dump 成 Hive，去進行后續的離線數倉的相關工作。切換成這套方案后，dump 操作就可以省掉，能夠做到流批的數據復用的能力來減少整體的中間存儲的成本。

第二個場景是多維分析場景，其特點是實時數據清洗后直接支持看板等實時的OLAP查詢。基于批流一體方案結合 Presto 查詢來滿足業務側分鐘級時延訴求，和秒級查詢響應訴求。我們團隊針對Presto進行了許多優化，包括native engine等相關技術，以實現高性能查詢。目前，該場景也在現場落地，并取得了不錯的收益。另外，因為整體的流批是數據表，存儲是統一的，所以不需要額外將其轉儲為Hive表，也不需要維護離線存儲的快照。

第三個場景就是批流復用的日志場景，一般大家直覺上會從 ODS 層切換做批流復用，字節內部，在實時場景會先對接埋點數據， Flink 端去做清洗，落到實時的存儲里面，然后對接看板等下游。在離線數倉上，也會將所有埋點信息存儲下來，根據具體的業務場景落成不同的 ODS 表，再去構建離線數倉的任務。

當我們整體把存儲換成 LAS 這套方案之后，只需要維護 ODS 層的數據，就能支持離線和實時兩個的場景去進行分析。

最后是飛書數倉的場景落地，整體鏈路比較清晰，分為實時和離線兩個鏈路，這里離線實時鏈路主要是去做一些人事信息變更之類的業務。離線鏈路主要是對一些長周期的問題去做一些數據的修正，把這部分修正的數據回補到我們的實時鏈路里面，讓下游的看板數據變得更準確。

在實時數據傳輸和離線數據處理兩個環節中，我們都采用了LAS之類的存儲來替換底層存儲。離線數據處理要求數據的處理時間在10-15分鐘內完成，因此用戶更慣用Spark處理離線數據處理環節。針對此，我們提供了基于Spark Thrift Server的解決方案，以減少離線數據處理中每個環節的資源申請開銷，維持常駐資源，讓用戶運行SQL來構建離線數據處理模塊。

在實時數據傳輸環節中，用戶原本采用Kafka構建基礎表并使用Hbase進行維表構建。因為Hbase對聯合主鍵的支持并不友好，用戶每次在讀寫時都需要去序列化和反序列化主鍵列，并將復合主鍵拼裝成單一主鍵，最后將其寫入Hbase中，然后再從Hbase中讀取。這樣的流程很復雜，且時間耗時較長。

那么除了替換之外，就像之前提到的一樣，我們會替換掉MQ或Kafka這樣的組件，并且我們還用其他存儲替代了Hbase。我們的主要目標是實現基于Flink的lookup join功能，并將小表直接加載到內存中，以提高lookup join的性能。因此，在實時鏈路這一塊，我們已經替換了所有的存儲組件。

四、未來規劃

最后分享一下未來規劃。首先，我們會探索更多業務場景，大規模落地更多模式。其次，在技術迭代方面，我們會對負載均衡和分離做出優化。負載分離在BTS內部（即內存服務內部）會針對一些小組件進行優化，比如將WAL文件刷入持久化存儲（如HDFS），這部分資源占用比較高，需要分離處理。另一方面，我們會更加細致化地對讀寫負載分離和內存服務負載均衡進行優化。此外，我們還會實現更精細的流批負載分離。第三點是查詢優化，我們會結合索引的能力，在內存層和整體存儲方案中通過構建索引優化塊的數據結構來加速查詢性能，包括點查和聯合查詢等。最后一點是與native engine的集成，以提升整體的讀寫速度。在這方面，我們需要對底層的log、block和parquet文件進行向量化處理。

五、Q&A環節

Q：Hudi支持流式的寫入與更新，那Kafka 是否可以被取代了？

A：我不確定是指社區的Hudi還是我分享的Hudi。所以我談一下我分享的整套方案，但我認為我們目前在某些方面還有欠缺。比較困難的是如何實現Kafka的exactly once語義。

Q：LAS支持的組件索引和二級索引是什么樣的索引結構？

A：實際上，這部分主要是社區原生實現，包括我們內部的實現也大部分已經貢獻到社區了。就組件而言，社區實現是基于哈希去進行分桶，并同時記錄一些布隆過濾器（Bloom filter）。關于二級索引，社區目前正在進行迭代，但并沒有完全合入。

Q：使用 BTS 可以加速 Flink 寫入 Hudi 的性能嗎？

A：會。時效性提升很多。因為第一個是我們寫內存，第二個是整個數據結構會相對 Hudi 來說會比較輕量一些。

Q：BTS 與 Hudi分別適用的應用場景。BTS 與Hudi的具體關系。

A：首先對于我們來說，我們整體的這套方案叫LAS， LAS 底層是基于 Hudi 去做的實現，為了支持流批一體存儲，我們在Hudi上加了一層內存緩存層BTS，結合查詢引擎，一整套方案稱之為 LAS。因為 BTS 是基于 Hudi 上架了一層，所以 BTS 整體的邏輯會跟 Hudi 強相關，它我兩者之間的交互主要就是內存文件到持久化存儲中間的一些交互，其他的大部分的設計會沿用Hudi的部分的邏輯，比方說我們會通過TSM, Table Service Management.去做一些比較優化的服務，比方說 Hudi 的compaction， BTS 的 WAL 的clean，就這些操作。

Q：LAS怎么保證查詢數據的準確性？

A：我們當前支持的語義是 At least once，就是說首先用戶能夠在數據里面加業務字段來判斷哪條數據是最新的At least once，但是這條數據有可能會被寫入多次，所以用戶下游需要做一些去重的操作。BTS 數據怎么保證一定是寫進來的？首先我們會寫 WAL 文件，就是 WAL 文件是持久化存儲上的文件，當文件寫完之后我們才會，我們會一邊寫 WAL文件一邊寫內存的數據，只有寫到 WAL 文件才會寫內存的數據，那整體都完成了之后才會返回給 client ACK，否則， client 會認為這次提交失敗了，會重新去往里寫，所以整體上一致性是不會有太大的問題的。