LAS 服務是什么？LAS 有哪些優化特性？本文將從基礎概念、數據庫內核特性優化、數據服務化、業務實踐等角度全方位介紹湖倉一體架構在LAS的探索與實踐。

LAS服務是什么？

在了解 Las 服務是什么之前，先來了解一下數據平臺整體行業的發展趨勢，大概分為三個階段。

第一階段，一般被稱為傳統數倉，一種從 1980 年開始的基于傳統數據庫技術來做的 BI 分析場景。在這種架構下，通常計算和存儲是高度一體的。整體系統能支撐的計算能力，依賴于服務提供商的硬件配置，整體成本高，存在物理上限，擴展起來比較麻煩。

第二階段，隨著技術的演進， 2010 年開始出現了以 Hadoop 技術體系為主流的傳統數據湖。在以 Hadoop 技術為主的數據平臺架構下，通常可以支持服務在普通硬件上面去部署，整體的計算和存儲的擴展性都得到了解決。基于開源技術生態，多個大型公司也參與到數據湖技術發展中來，整體生態繁榮度也在逐步提升。

但在這一階段凸顯出了一個問題，隨著生態技術的發展，越來越多的開源組件開始累積。對于一個企業來說，為了解決不同領域的問題，需要運維多個開源的組件，來滿足不同領域的數據需求，就導致整個企業的技術運維成本逐步提升。

基于這個問題，隨著技術的進一步發展，在 2020 年，湖倉一體的架構開始被提出。

相比起傳統數據湖，湖倉一體架構支持原生的 ACID 能力，支持像 BI 分析、報表分析，機器學習和流式分析多種類型的計算范式，以及云上的對象存儲和彈性計算能力。以上能力，讓湖倉一體架構能夠有效地去解決企業的對數據規模，以及對計算能力的彈性伸縮需求。同時，湖倉一體可以在很大程度上規避傳統 Lambda 架構存在的多個計算組件，或者多種架構范式導致的架構負擔，讓企業能夠更專注地去解決他們的業務價值。

LAS 就是基于湖倉一體的架構進行設計的。從上圖來看，LAS 架構整體上分為三個部分。最上層是開發工具層，開發工具層會通過計算層提供的統一 SQL 訪問服務去訪問計算層，根據用戶的 SQL 類型自動做 SQL 解析。所有引擎計算能力統一由彈性容器服務來提供，可以支持彈性伸縮，按需使用。

再往下就是湖倉一體的存儲層。首先，湖倉一體存儲會通過統一的元數據服務，向計算層提供統一的元數據視圖，屏蔽底層的具體元數據實現細節，可以使多個引擎無縫對接到統一的元數據服務。

接下來是湖倉存儲引擎，它主要提供了事務管理能力，也就是 ACID 的能力，以及對數據批流一體的讀寫能力。

再往下就是 LAS 基于火山引擎對象存儲服務 TOS 和 CloudFS ，來提供 EB 級的數據存儲能力和數據訪問的緩存加速能力。

以上就是 LAS 整體的技術架構。

LAS 數據湖內核剖析

這一版塊將向大家呈現 LAS 數據湖內核的特性及優化。

LAS 的數據湖內核—— ByteLake 它是什么？

首先，ByteLake 是基于開源 Apache Hudi 進行內部增強的湖倉一體存儲引擎，提供湖倉一體的存儲能力。

它的第一個主要能力是提供了湖倉統一的元數據服務，完全兼容開源的 Hive Metastore，可以無縫對接多種計算引擎。第二個主要能力是可以支持對海量數據的 Insert，完全兼容 Hive SQL，可以平遷傳統數倉場景下的 Hive 任務。第三，ByteLake 支持對大規模歷史數據的 Update 和 Delete，以及對新增數據的 Upsert 和 Append 能力。最后，ByteLake 支持流批一體的讀寫能力，提供流式讀寫的 source 和 sink，支持近實時分析。

ByteLake 又是怎么做到這些能力的呢？接下來從以下幾個特性來展開闡述。

如何實現高效數據更新？

第一個場景是流式寫入更新場景。在這種場景下，最明顯的特點就是小批量數據頻繁寫入更新。但主要的問題是如何去定位要寫入的記錄呢？是做 update 操作還是 insert 操作？

在這樣的背景下，ByteLake 提供了一種 Bucket Index 的索引實現方案。

這是基于哈希的一種索引實現方案。它可以快速地去定位一條記錄所對應的 Fail Group，從而快速定位當前記錄是否已經存在，來判斷這一條記錄是做 Update 還是做 Insert 操作，從而可以快速地將這種小規模的數據去添加到 Append Log。在讀取時，通過 Compaction 就可以將 LogFile 和 BaseFile 里邊的數據進行 Merge 去重，從而達到數據更新的效果。

針對日志數據入湖，通常來說是不需要主鍵的，這種基于 Hash 索引的實現方式，是需要有 Shuffle 操作的。因為在基于 Hash 的索引實現中，當一批數據過來之后，會根據這一批數據去找分別對應的 File Group，再基于 File Group 去聚合要更新的這些數據，通過同一個 Task，去更新同一個 File Group 來實現原子寫入。

在數據 Shuffle 的過程，其實對于數據湖日志寫入是有額外的開銷的，但 ByteLake 提供了一種 Non index 的實現方案，去掉了索引的約束，可以減少數據 Shuffle 的過程，從而達到快速入湖的能力。

存量數據如何高效更新?

存量數據，一大特點就是數據量大，單表的規模可能有幾百 TB ，甚至到 PB 的級別。針對于這種大規模的歷史數據的更新場景，如何去提升更新性能？其實最主要的就是要如何去降低數據更新的規模。

基于此，ByteLake 提出了一種實現方案——Column Family，將單表多列的場景分別存儲到不同列簇。不同的文件可以基于 Row Number 進行聚合，合并后就是一個完整的行。如果要更新歷史數據，只需要去找到要更新的那些列對應的 Column Family 對應的文件，把這些文件做一些局部更新，就可以達到整體更新的效果。從而在很大程度上減少這些非必要數據的掃描，提升存量歷史數據更新場景的性能。

如何提升并發性能？

談到并發，通常會有兩部分內容。比如有很多個任務同時去往 ByteLake 引擎里邊寫數據，這就意味著有大批量的任務去訪問 ByteLake 的 MetaStore Service。在這種場景下，ByteLake MetaStore Service 就會成為一個性能瓶頸。

為了突破這個瓶頸，除了無限的堆加資源之外，另一個比較有效的方案就是增加緩存。通過元數據服務端去緩存比較熱點的數據，比如 Commit Metadata 和 Table Metadata，來達到服務端的性能提升。

另外一塊，是在引擎側做優化。比如在 Flink 引擎層面將 Timeline 的讀取優化到 JobManager 端。同一個任務下，只要 JobManager 去訪問 Hive ByteLake MetaStore Service，緩存到 JobManager 的本地之后，所有的 TaskManager 只要去訪問 JobManager 本身緩存的 Timeline 信息就可以了。

從單個任務的視角來看，比如多個任務要同時去更新同一張表，這種情況下要保證數據的正確性，同時又能保證并發性能，應該如何來做？ByteLake 提供的解決方案——基于樂觀鎖的一個并發控制。

針對多任務寫同一個表的場景，ByteLake 可以支持多種并發策略的設置。業務可以根據對數據一致性的要求，以及對數據并發性能的要求，選擇靈活的并發策略，來達到它的數據并發寫入的性能指標。

LAS 數據湖服務化設計

這個版塊將向大家呈現 ByteLake 服務化過程中的一些設計實踐。

CatalogService ：統一的元數據視圖

CatalogService 主要提供了與 HMS 的兼容接口，同時為所有的查詢引擎提供了統一的元數據視圖，解決了異構數據源的元數據管理問題。

CatalogService 整體分三層，第一層是 Catalog Federation，提供統一的視圖和跨地域的數據訪問能力。以及提供了對源數據請求的路由能力，可以根據元數據請求的類型，支持通過 Mapping 的方式，來路由不同的服務請求對應的底層元數據服務實例。

第二層是 CatalogService 下層的具體元數據服務的實現，比如 Hive MetaStore Service 以及 ByteLake MetaStore Service 等。可能還有不同的元數據服務對接到 CatalogService，來統一向上層引擎提供這種元數據服務。

最后一層是 MetaStore 的存儲層，它通過插件式的方式來提供不同的存儲引擎，來滿足上層不同元數據服務實例的存儲要求。

BMS 詳解

湖倉一體元數據管理服務

Bytelake MetaStore Service，簡稱 BMS，它是一個湖倉一體的元數據管理服務，整體的架構分為以下幾個部分。首先第一個就是 Catalog，Catalog 是對單表的元數據訪問的抽象。主要邏輯是通過 MetaStore Client 來訪問 Meta Server，同時它會去緩存單表的 Schema 信息以及屬性等信息。

另外一部分就是 Meta Server，也就是 BMS 里邊最核心的部分。它主要是包含兩大部分服務層，第一是 Bytelake MetaStore 元數據服務模型，比如 Table Service，Timeline Service，Partition Service 和 Snapshot Service。存儲層提供了 MetaStore 所有元數據的存儲能力。最后一部分就是 Eventbus， Eventbus 主要目的是為了將元數據的 CUD 事件發送給監聽者，來達到元數據信息的分發和同步。

元數據寫入流程

關于元數據寫入流程，簡單來講，當有一個 Client 去提交了 Instant 之后，Bytelake Catalog 會去訪問 Bytelake Meta Store 的接口，會將 Instance 改成 Completed，然后將請求發到 Bytelake 的 MetaStore，之后 Bytelake MetaStore Server 會做一個原子提交。

在此之后，Timeline Service 會把提交的狀態更新到數據庫里邊。接下來這些分區信息將再被提交給 Partition Service，同步到對應的分區存儲表里去。最后一步，把這些所有的變更作為一個快照，同步到 Snapshot Service 里，它會把文件層面的變更存儲到數據庫里，做持久化存儲。

元數據讀取流程

對于源數據的讀取流程，舉個例子，有一個計算引擎它讀取了一個 SQL，通過 SQL 解析拿到一張表，這張表會通過Bytelake Catalog Service去請求Bytelake MetaStore，最終會路由到 Table Service 拿到這些表的信息。

拿到表的信息做 SQL Plan 優化的時候，會做一些分區的下推或裁剪。這個時候會去請求到 Bytelake 的 Partition Service 做過濾，接著會根據分區信息去掃描文件，在此過程中會去請求 Timeline Service 獲取對應的 Timeline 信息。接下來，基于 Timeline 的信息時間去 Snapshot Service 拿到對應文件，再通過 SQL 執行器來實現數據文件的讀取。

元數據變更通知

元數據變更通知具體的實現流程主要依托于兩個部分。

一是 Eventbus，二是 listener。所有的元數據請求都會發送到 Eventbus，由 Eventbus 分發事件到所有已經注冊的 Listener 上面。listener 再根據下游系統的需求，去訂閱 Eventbus 里邊的對應事件類型進行響應，從而達到讓上下游的組件感知到元數據的變化，實現元數據的同步。

TMS 詳解：

統一表管理服務

LAS 的另外一個服務——TMS，全稱是 Table Management Service。它主要解決的問題是異步任務的托管優化。為什么會做異步任務的托管優化？因為正常來講，Flinker SQL 任務寫 ByteLake 表的過程，其實就是把批量的數據寫入下游表里邊去。隨著時間的推移，一個是 Commit 的日志非常多，另外一個是小文件非常多。通常的 Flink 引擎層面的實現方案，是在數據寫了一定的次數后，追加一個 Compaction 操作，把之前寫入的文件做一個壓縮。

但針對流式任務去做 Compaction，對正常的流式任務穩定性有很大影響，因為壓縮本身是一個開銷比較大的動作，對流式計算資源的消耗是很難去評估的，會導致整個流式寫入任務的波動，從而影響流式寫入任務的穩定性。

基于此，LAS 提供了一個統一的表管理服務，異步托管這些本身內置到引擎內部的任務，統一由 Table Management Service 來托管。它整體的架構是一個主從架構，主要包含的組件一個是 Event Receiver，用來接收 Metastore 下發的一個 Event。PlanGenerator 就是根據 Meta store Server 下發的 Event 信息，來觸發 Action Plan 的生成。

什么是 Action Plan？簡單講，就是這一次要做哪些事情，比如你要做一個壓縮任務，還是做一次歷史文件的清理，還是做一些小文件的合并，都稱為 Action Plan。Job Scheduler 就是去調度需要被執行的 Acting Plan。

什么是 Job Manager？它主要用于和集群交互，比如 Yarn 或 K8S，管理 Action Plan 對應的執行任務，做一些任務運維層面的工作。

執行計劃生成

就執行計劃生成展開來講，Plan Generator 會接收 Metastore 下發的一些事件，根據用戶在表的 DDL 里的配置策略，來決定是否要生成執行計劃。

這個策略通常會有幾種，比如，一種基于它 Delta Commit 的數量，連續提交了多次達到了一定的閾值，就會觸發一個 Action Plan 的生成，來做一次數據的壓縮。另外一種，是根據 Log File 的大小，來判斷 Compaction 操作是否需要執行。PlanGenerator 策略會根據當前 Log File 的 Meta 信息，來決定是否要觸發 Action Plan 的生成。

執行計劃調度管理

執行計劃生成結束之后，最后一步就是怎么去調度管理執行計劃。執行計劃調度的核心流程主要由 Job Scheduler 來做，Job Scheduler 會定時地去輪詢已經生成的 Action Plan，再分發給 Job Manager。Job Manager 拿到了 Action Plan 之后，會到集群上提交一個任務，同時不斷去輪詢任務的狀態，更新任務的狀態到數據庫，保證 Action Plan 執行的可靠性和穩定性。通常 JobScheduler 一般會有先進先出的調度策略，來保證 Action Plan 達到預期調度效果。

LAS 在字節跳動的業務實踐?

抖音電商在湖倉一體架構下的業務實踐

抖音電商的業務場景，主要是營銷大促、流量診斷以及物流狀態的監控。他們的業務痛點是什么？數據量大，計算邏輯復雜，同質數據源也比較多，寬表的構建成本比較高，包括一些其他的技術問題。還有一個痛點就是計算周期長，增量計算成本比較高。

基于 LAS 湖倉一體架構下，可以解決哪些問題呢？

首先，通過 LAS 快數據入湖能力，可以解決多數據源的快速入湖。把外部的業務系統和業務日志，通過 LAS 這種實時入湖能力快速導入到 ODS 層。通過離線數倉可以直接引用 ODS 層的準實時入庫數據，來達到離線數倉的日增量數據，同步提升數據的時效性。

其次，實時數倉中 DW 層的一些明細數據，也可以通過流式入湖的能力，直接導入到 ByteLake，達到數據復用的目的。當把這些數據導到了 ByteLake 之后，針對大寬表場景，就可以基于 ByteLake 的多流拼接能力，直接在底層的存儲引擎層，實現寬表的構建。從而解決在常規場景下，通過 Flink SQL 做多源或多流 join，導致的任務狀態比較大，或者任務處理復雜度比較高的這種穩定性問題，從而更好地去保障業務數據的及時性和穩定性。

消費行業傳統數倉架構升級

消費行業的客戶場景，實際就是在零售場景下的財務管理、庫存管理相關的一些計算場景。客戶的實現方案基于傳統的數據庫，業務和離線分析的請求都是統一在一個傳統數據庫上邊來做的。

在這種場景下，其實整個 RDBMS 要同時承接業務處理邏輯和離線 ETL 分析邏輯。隨著業務數據量的增長，很快就會發現傳統數據庫的計算能力和存儲支撐能力達到了上限，導致計算能力不足，擴展性比較差，無法在滿足后續的業務數據規模的上量。

LAS 針對這種場景的解決方案，是將客戶的離線 ETL 的分析場景，通過實時集成的方式直接導入到 LAS 里邊，通過 LAS 的彈性計算能力，為用戶的 ETL 分析場景提供有效的算力保障。在滿足客戶低成本約束的情況下，達到客戶預期的計算效果，和對數據產出的及時性的要求。同時會通過云上的 ByteHouse 服務來解決客戶自建的 CK 的運維成本以及性能調優的問題。優化了原有的基于 RDBMS 的數據鏈路，保證業務數據量快速增長的同時，滿足它的底層的算力要求。

湖倉一體架構下的批流融合計算

典型場景就是數據實時入湖，客戶的數據源會通過 Flink SQL 持續地去寫入到 LAS 的 Bytelake 表里。但下游如果是一個離線任務，其實用戶沒辦法很便利地去判斷數據寫到了哪個位置，或者分區數據現在是不是已經完備的。

如果僅依賴系統時間來實現，比如在上游的這種 Flink SQL 任務，在寫入過程正常時倒沒有特別大的問題。但是一旦上游 Flink SQL 任務出現一些數據積壓或者任務異常的場景，下游依賴系統時間去調度，就會存在某些分區會出現數據空洞或數據偏移的問題。例如本來數據應該落在 7 點的分區，因為上游的這些 SQL  任務的消費延遲，導致 7 點的數據并沒有準時地落下來, 導致下游去消費 7 點的數據的時候，拿到的是一個不完整的數據，導致出現數據空洞或數據偏移的問題。

針對這種場景，LAS 提供了一種叫歸檔的能力，也就是在 Flink SQL 寫入的過程中，會基于業務事件時間實時寫入對應的數據分區。通過 ByteLake 提供歸檔能力，分區數據就緒后，可自動生成一個歸檔標簽。下游的 spark SQL 任務可以根據分區是否有歸檔標簽，來判斷對應分區的數據是否就緒，來決定當前離線任務是不是要調度起來。

這項能力的實現邏輯，其實就是 Flink SQL 每次去提交一個 Commit 的時候，會去判斷當前提交的業務的事件時間，是否比當前的未提交分區的時間超過了某一個閾值。比如當前分區的時間是 7 點，Flink SQL 在持續提交微批數據的時候，它判斷出來當前的最小的業務時間已經到 7 點半了，而業務定義的可容忍的延遲間隔是 15 分鐘， ByteLake 認為這個數據其實已經寫完了，就會把 7 點的分區數據打上一個歸檔標簽，來標示數據已經完成了。下游就可以去正常地去消費 7 點的分區數據，從而保證數據的完整性。

在提供了這種歸檔能力的情況下，LAS 的整體計算鏈路就可以實現批流融合。比如 ODS 的 ByteLake 表是一個準實時的表，下層的 Spark SQL 任務可以直接通過 Spark ETL 去做處理，產出一個離線表。可能后邊還會有一些 SQL 場景依賴離線表做數據的準實時消費。在這種情況下，Flink SQL 會再生成一張 ByteLake 表，這張表同樣可以被下游的 Spark SQL 的離線任務依賴，從而達到在整個 Pipeline 里，做到批流計算相互融合的狀態。