一、背景與挑戰(zhàn)

首先來(lái)介紹一下相關(guān)背景。

傳統(tǒng)數(shù)倉(cāng)存在實(shí)時(shí)和離線(xiàn)兩條鏈路，來(lái)滿(mǎn)足業(yè)務(wù)對(duì)于時(shí)效數(shù)據(jù)的時(shí)效性和數(shù)據(jù)量的不同需求。離線(xiàn)會(huì)維護(hù)歷史的全量視圖，實(shí)時(shí)會(huì)維護(hù)增量視圖，最后在服務(wù)層去進(jìn)行數(shù)據(jù)的匯總，從而支持后續(xù)的在線(xiàn)的serving、 OLAP 查詢(xún)以及看板的應(yīng)用等等。 

因?yàn)樘幚韴?chǎng)景的差異，在實(shí)時(shí)和離線(xiàn)數(shù)倉(cāng)的具體實(shí)現(xiàn)上，依賴(lài)的底層存儲(chǔ)計(jì)算引擎基本上是完全隔離的，實(shí)時(shí)依賴(lài)的主要是以 Flink 為代表的流式計(jì)算引擎來(lái)做計(jì)算，而離線(xiàn)依賴(lài)主要是以 Spark 為代表的引擎，實(shí)時(shí)主要依賴(lài) KV 或 MQ 這樣的多種存儲(chǔ)選型。離線(xiàn)則常常采用 Hive 為代表的存儲(chǔ)引擎，傳統(tǒng)的數(shù)倉(cāng)架構(gòu)，它本質(zhì)上結(jié)合了流計(jì)算和 批計(jì)算的優(yōu)勢(shì)，通過(guò)兩套代碼來(lái)兼容實(shí)時(shí)數(shù)據(jù)和離線(xiàn)數(shù)據(jù)的優(yōu)勢(shì)，彌補(bǔ)各自的缺點(diǎn)。但是這兩套架構(gòu)或代碼也帶來(lái)了兩倍的資源成本，并且因?yàn)榈讓佑?jì)算引擎的不同，對(duì)于相同算子的處理的語(yǔ)義不是完全一致的，它們的計(jì)算結(jié)果就會(huì)存在差異。所以對(duì)于研發(fā)同學(xué)來(lái)說(shuō)，這部分差異也會(huì)給數(shù)據(jù)校驗(yàn)等其它一些工作帶來(lái)額外的負(fù)擔(dān)?？偨Y(jié)下來(lái)，傳統(tǒng)數(shù)倉(cāng)的 Lambda 架構(gòu)主要存在三個(gè)問(wèn)題要解決：第一個(gè)是一套計(jì)算邏輯，但需要寫(xiě)離線(xiàn)和實(shí)時(shí)兩套代碼，帶來(lái)了兩倍的運(yùn)維成本；第二是離線(xiàn)實(shí)時(shí)兩條鏈路帶來(lái)了資源冗余的問(wèn)題，雙倍的資源的成本；第三是兩套引擎計(jì)算口徑不完全對(duì)齊，導(dǎo)致數(shù)據(jù)校準(zhǔn)方面會(huì)有比較大的困難。所以業(yè)內(nèi)提出了希望通過(guò)批流一體來(lái)解決當(dāng)前傳統(tǒng) Lambda 架構(gòu)的問(wèn)題。基于上述問(wèn)題，再結(jié)合內(nèi)部的實(shí)際場(chǎng)景，批流一體的訴求可以分為兩種，第一種是計(jì)算的批流一體，第二種是存儲(chǔ)的批流一體。計(jì)算的批流一體指的是希望通過(guò)底層一套系統(tǒng)，業(yè)務(wù)層的一套代碼同時(shí)滿(mǎn)足離線(xiàn)和實(shí)時(shí)的開(kāi)發(fā)需求，從而解決兩套系統(tǒng)帶來(lái)的研發(fā)效率、人工成本、運(yùn)維成本等資源成本相關(guān)的問(wèn)題。另外我們希望一套系統(tǒng)能夠?qū)R計(jì)算指標(biāo)的口徑來(lái)解決數(shù)據(jù)一致性的問(wèn)題。

存儲(chǔ)的流批一體包括，第一是實(shí)時(shí)和離線(xiàn)所在的存儲(chǔ)統(tǒng)一，第二是實(shí)時(shí)和離線(xiàn)的數(shù)據(jù)能夠復(fù)用。實(shí)時(shí)和離線(xiàn)存儲(chǔ)統(tǒng)一是指我們希望實(shí)時(shí)和離線(xiàn)能夠使用統(tǒng)一選型的存儲(chǔ)，這就要求存儲(chǔ)能夠滿(mǎn)足大規(guī)模、全量和增量數(shù)據(jù)的讀寫(xiě)訴求。批流數(shù)據(jù)復(fù)用指的是批處理能夠使用流處理的結(jié)果數(shù)據(jù)，提升整個(gè)離線(xiàn)數(shù)倉(cāng)的產(chǎn)出時(shí)間。典型的 case 就是 ODS 層的數(shù)據(jù)復(fù)用，另外流處理也能夠復(fù)用批處理的數(shù)據(jù)來(lái)解決鏈路冷啟動(dòng)的時(shí)候，需要將離線(xiàn)的數(shù)據(jù)回灌到實(shí)時(shí)存儲(chǔ)中的額外成本，我們通過(guò) LAS 去實(shí)現(xiàn)了批流一體的能力。在計(jì)算層，我們的解決思路是對(duì)外暴露統(tǒng)一的 SQL，底層根據(jù) SQL 處理的場(chǎng)景選擇不同的執(zhí)行引擎去執(zhí)行，對(duì)用戶(hù)屏蔽以底層執(zhí)行引擎的差異性。之所以這樣做的原因是我們認(rèn)為不同引擎適用于不同場(chǎng)景，很難找到引擎能夠同時(shí)滿(mǎn)足實(shí)時(shí)和離線(xiàn)，不同時(shí)效性和數(shù)據(jù)規(guī)模的要求，在 SQL 層，我們對(duì)齊了底層執(zhí)行引擎的差異性來(lái)實(shí)現(xiàn)計(jì)算口徑的對(duì)齊，解決了上面說(shuō)到的一致性問(wèn)題。在存儲(chǔ)層，我們基于湖倉(cāng)一體的架構(gòu)，通過(guò)數(shù)據(jù)湖實(shí)現(xiàn)了批流一體存儲(chǔ)的能力。除了能夠支持流式的增量和批式的全量讀寫(xiě)之外，我們還支持了高效的 OLAP 查詢(xún)能力以及維表 join 的能力。

既然提到了LAS，我們看一下 LAS 的整體結(jié)構(gòu)， LAS全稱(chēng)是 Lakehouse Analysis Service。湖倉(cāng)一體的數(shù)據(jù)分析服務(wù)，融合了湖與倉(cāng)的優(yōu)勢(shì)，既能夠利用湖的優(yōu)勢(shì)將所有的數(shù)據(jù)都存儲(chǔ)到廉價(jià)存儲(chǔ)中，供積極學(xué)習(xí)、數(shù)據(jù)分析等場(chǎng)景使用，又能夠基于數(shù)據(jù)湖構(gòu)建數(shù)倉(cāng)，讓BI、報(bào)表等業(yè)務(wù)場(chǎng)景去使用。 

LAS具有如下一些特性：首先是能夠支持統(tǒng)一的元數(shù)據(jù)，避免在數(shù)據(jù)湖中存在數(shù)據(jù)孤島的問(wèn)題，每個(gè)數(shù)據(jù)都是可追溯的。第二個(gè)是依托數(shù)據(jù)湖提供ACID 的能力。第三點(diǎn)是機(jī)器支持企業(yè)級(jí)的權(quán)限管控。第四點(diǎn)是支持資源的極致彈性擴(kuò)縮容，降低用戶(hù)的使用成本。最后是引擎的內(nèi)核的極致優(yōu)化，提供高效的讀寫(xiě)性能。LAS 整體架構(gòu)，最上面一層是湖倉(cāng)開(kāi)發(fā)工具，為數(shù)據(jù)應(yīng)用場(chǎng)景提供能力支撐。下面一層是數(shù)據(jù)分析引擎，支持流批一體SQL，解決計(jì)算批流一體的問(wèn)題，并且支持根據(jù) SQL 的特點(diǎn)去做引擎的智能選擇和加速。針對(duì) OLAP 分析，我們會(huì)將 SQL 路由到不同的執(zhí)行引擎去執(zhí)行，比如對(duì) Ad-hoc我們會(huì)用 Presto 去進(jìn)行查詢(xún)。再往下一層是統(tǒng)一元數(shù)據(jù)層，最后一層是基于 Hudi 去實(shí)現(xiàn)的流批存儲(chǔ)層。本文會(huì)聚焦在流批一體存儲(chǔ)的細(xì)節(jié)實(shí)現(xiàn)上。

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

在構(gòu)建實(shí)時(shí)數(shù)倉(cāng)新鏈路的時(shí)候，對(duì)于鏈路冷啟動(dòng)，需要使用歷史分區(qū)的數(shù)據(jù)，所以我們需要將離線(xiàn)的數(shù)據(jù)回灌到實(shí)時(shí)鏈路的 MQ 里面，受限于 MQ 帶寬的限制，整體的回溯周期可能會(huì)非常的長(zhǎng)，并且操作很復(fù)雜。另外當(dāng)計(jì)算指標(biāo)有問(wèn)題，或者是整體的計(jì)算口徑需要調(diào)整的時(shí)候，也需要使用離線(xiàn)的數(shù)據(jù)去對(duì)實(shí)時(shí)數(shù)據(jù)進(jìn)行回刷，同樣它也會(huì)遇到回溯周期長(zhǎng)，操作復(fù)雜等一系列問(wèn)題。通過(guò)介紹可以看出實(shí)時(shí)數(shù)據(jù)倉(cāng)庫(kù)整體相對(duì)比較復(fù)雜，存儲(chǔ)方式和構(gòu)建標(biāo)準(zhǔn)沒(méi)有完全統(tǒng)一。為了更精細(xì)地分析實(shí)時(shí)數(shù)據(jù)倉(cāng)庫(kù)對(duì)于流批一體存儲(chǔ)的需求，我們基于數(shù)據(jù)量延遲、數(shù)據(jù)一致性要求和計(jì)算周期等維度，將場(chǎng)景劃分成了三類(lèi)：日志計(jì)算、長(zhǎng)周期計(jì)算和全量計(jì)算。

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

二、設(shè)計(jì)方案

接下來(lái)看一下我們的流批一體存儲(chǔ)方案，結(jié)合剛剛討論的流批一體存儲(chǔ)目標(biāo)，我們發(fā)現(xiàn)現(xiàn)有的基于數(shù)據(jù)湖倉(cāng)一體的架構(gòu)，實(shí)際上已經(jīng)可以滿(mǎn)足大多數(shù)要求了。當(dāng)前數(shù)據(jù)湖倉(cāng)一體的架構(gòu)已經(jīng)支持所有數(shù)據(jù)入湖，并支持Spark、Flink引擎，同時(shí)也可以進(jìn)行離線(xiàn)和實(shí)時(shí)的數(shù)據(jù)操作。在下游數(shù)據(jù)應(yīng)用方面，數(shù)據(jù)湖倉(cāng)一體的架構(gòu)還支持ihook、metastore、adhoc等OLAP查詢(xún)方式。在字節(jié)內(nèi)部使用的場(chǎng)景中，業(yè)務(wù)會(huì)通過(guò)Flink實(shí)時(shí)將數(shù)據(jù)入湖，使用Spark批量回溯更新湖內(nèi)的數(shù)據(jù)，并且下游會(huì)使用Presto查詢(xún)服務(wù)來(lái)觸及下游的看板。因此，我們信息湖倉(cāng)庫(kù)主要使用了Hudi這樣的開(kāi)源方案。在功能方面，數(shù)據(jù)湖倉(cāng)庫(kù)基本符合了實(shí)時(shí)和離線(xiàn)數(shù)倉(cāng)對(duì)于流批一體存儲(chǔ)的需求，而這主要是因?yàn)镠udi本身提供了事務(wù)支持，我們?cè)趦?nèi)部還進(jìn)行了桶索引機(jī)制的優(yōu)化以進(jìn)一步提高入湖的性能，并且通過(guò)metastore的元數(shù)據(jù)服務(wù)來(lái)支持并發(fā)寫(xiě)入功能。此外，Hudi原生支持多引擎，因此既可以對(duì)批流進(jìn)行讀寫(xiě)消費(fèi)，也可以使用Presto進(jìn)行交互式分析。

在內(nèi)部，湖倉(cāng)一體架構(gòu)大規(guī)模地落地了離線(xiàn)的數(shù)倉(cāng)場(chǎng)景和部分近實(shí)時(shí)的數(shù)倉(cāng)場(chǎng)景。但是因?yàn)?Hudi 本身的或者數(shù)據(jù)庫(kù)本身分鐘級(jí)別的可見(jiàn)性，它還是沒(méi)有辦法做到實(shí)時(shí)數(shù)倉(cāng)存儲(chǔ)的標(biāo)準(zhǔn)方案。

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

上層服務(wù)層主要分為兩個(gè)組件， BTS 和 Table Service Management。BTS 是基于內(nèi)存構(gòu)建的服務(wù)層，它主要是來(lái)解決實(shí)時(shí)場(chǎng)景下數(shù)據(jù)讀寫(xiě)的時(shí)效性的問(wèn)題，通過(guò)內(nèi)存去對(duì)數(shù)據(jù)讀寫(xiě)進(jìn)行加速，TSM (Table Service Management)，是表優(yōu)化的服務(wù)，它會(huì)異步地去執(zhí)行一些表優(yōu)化的操作，從而實(shí)現(xiàn)對(duì)查詢(xún)的加速。

這里的表優(yōu)化操作指的包括社區(qū)原生的壓縮聚合 clustering，以及一些索引異步構(gòu)建，視圖異步構(gòu)建的一些操作。壓縮聚合指的是對(duì)日志文件和基礎(chǔ)文件進(jìn)一步合并以生成新的列式存儲(chǔ)文件，這樣對(duì)整體查詢(xún)效率而言更優(yōu)。而 clustering 則是合并小文件以減少文件開(kāi)銷(xiāo)。當(dāng)前 TSM 只支持這兩種能力以及清理能力，我們計(jì)劃結(jié)合社區(qū)現(xiàn)有的 MDT 能力來(lái)異步構(gòu)建多級(jí)索引，以提升交互式查詢(xún)的性能。

表優(yōu)化操作是一個(gè)完全異步的過(guò)程。這部分是我們自主開(kāi)發(fā)的服務(wù)，因?yàn)橐恍┥鐓^(qū)原生實(shí)現(xiàn)并沒(méi)有做到完全異步。為什么要異步呢？因?yàn)?compaction 和 clustering 的執(zhí)行時(shí)間比較長(zhǎng)，同步操作會(huì)影響數(shù)據(jù)湖的寫(xiě)入速度，特別是在實(shí)時(shí)場(chǎng)景下不可接受。而社區(qū)的異步操作僅指寫(xiě)入時(shí)不阻塞，但是 compaction 會(huì)共享寫(xiě)入資源在同一個(gè)應(yīng)用程序內(nèi)執(zhí)行。這可能會(huì)影響寫(xiě)入作業(yè)的穩(wěn)定性，因此我們?cè)趦?nèi)部落地過(guò)程中發(fā)現(xiàn)了這個(gè)問(wèn)題，最后實(shí)現(xiàn)了一個(gè)完全異步的調(diào)度執(zhí)行，同時(shí)不共享寫(xiě)入資源的服務(wù)。在具體的執(zhí)行層面上，我們還利用混部的資源以降低成本。

1、數(shù)據(jù)組織形式

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

2、數(shù)據(jù)讀寫(xiě)方式

再來(lái)看一下整個(gè)數(shù)據(jù)讀寫(xiě)的交互方式。首先，做了流批復(fù)雜的分離，因?yàn)榱鲌?chǎng)景和批場(chǎng)景對(duì)于數(shù)據(jù)的可見(jiàn)性和時(shí)效性的要求是不完全一致的。對(duì)于批量回溯的場(chǎng)景，用戶(hù)并不是希望能夠馬上可見(jiàn)，只是希望把這部分?jǐn)?shù)據(jù)做好更新和校準(zhǔn)而已。

在批量數(shù)據(jù)更新的場(chǎng)景中，數(shù)據(jù)會(huì)直接寫(xiě)入持久性存儲(chǔ)中，也就是會(huì)寫(xiě)入到HDFS上，而不是通過(guò)內(nèi)存。這種方法可以極大地提高我們的讀寫(xiě)吞吐量。對(duì)于流式讀寫(xiě)的情況，會(huì)首先訪(fǎng)問(wèn)BTS之類(lèi)的內(nèi)存服務(wù)進(jìn)行讀寫(xiě)。這里的主要實(shí)施細(xì)節(jié)是，在寫(xiě)入數(shù)據(jù)時(shí)，我們會(huì)優(yōu)先寫(xiě)入內(nèi)存中，會(huì)首先寫(xiě)WAL文件以保障容錯(cuò)。在讀取數(shù)據(jù)時(shí)，由于內(nèi)存中的數(shù)據(jù)不會(huì)一直存在，因?yàn)閏ache會(huì)定期清理，所以讀取時(shí)會(huì)優(yōu)先訪(fǎng)問(wèn)內(nèi)存中的數(shù)據(jù)。如果發(fā)現(xiàn)內(nèi)存中沒(méi)有數(shù)據(jù)，我們會(huì)先加載WAL文件并對(duì)其進(jìn)行預(yù)加載，以盡可能地將數(shù)據(jù)優(yōu)先加載到內(nèi)存中，以保證流式讀取的時(shí)效性。如果發(fā)現(xiàn)WAL文件也不存在，或者被清理了，那么我們就會(huì)轉(zhuǎn)而去讀取持久性存儲(chǔ)中的日志。在大多數(shù)情況下，這是流式讀寫(xiě)，整個(gè)周期相對(duì)較短。因此，在內(nèi)存中，WAL文件的存儲(chǔ)能夠做到一周之內(nèi)的時(shí)效性。一周之內(nèi)的用戶(hù)都可以正常從內(nèi)存中消費(fèi)這部分?jǐn)?shù)據(jù)。如果用戶(hù)希望存儲(chǔ)長(zhǎng)周期的數(shù)據(jù)，那么他可能需要承擔(dān)更多的存儲(chǔ)成本。我們需要盡可能避免從HDFS上加載日志文件，這是整體的數(shù)據(jù)讀寫(xiě)方式。

3、BTS架構(gòu)

剛剛提到的流讀的場(chǎng)景，我們也做了讀寫(xiě)的負(fù)載分離，會(huì)有單獨(dú)的讀集群去承接整體的讀流量，來(lái)避免它影響寫(xiě)入節(jié)點(diǎn)的性能。我們來(lái)看一下整體的BTS架構(gòu)，BTS首先是Master-Slave架構(gòu)。Master主要負(fù)責(zé)一些元數(shù)據(jù)信息的管理，它的結(jié)構(gòu)與HDFS相似。Table Server以Slave的形式存在，負(fù)責(zé)數(shù)據(jù)的讀寫(xiě)，并在其上存儲(chǔ)由若干個(gè)block組織而成的文件。對(duì)于Master，它管理的元數(shù)據(jù)包括Table Server的信息以及block的元數(shù)據(jù)管理。由于元數(shù)據(jù)管理的需要，它必然會(huì)引入一定的負(fù)載均衡機(jī)制。因此，我們目前實(shí)現(xiàn)了比較簡(jiǎn)單的負(fù)載均衡機(jī)制，旨在避免某一Table Server內(nèi)存被打爆的情況。

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

三、落地場(chǎng)景

我們的主要落地場(chǎng)景是流式數(shù)據(jù)計(jì)算，它類(lèi)似于離線(xiàn)數(shù)倉(cāng)，需要進(jìn)行一些ETL清理和簡(jiǎn)單的聚合計(jì)算。左側(cè)是整體架構(gòu)方案，我們使用了基于Hudi加BTS的數(shù)據(jù)湖方案來(lái)替換MQ，從而實(shí)現(xiàn)每一層數(shù)據(jù)表的存儲(chǔ)。在維表上，我們目前仍使用KV存儲(chǔ)。我們當(dāng)前的目標(biāo)是替換MQ場(chǎng)景。

原先離線(xiàn)需要將實(shí)時(shí)表從 MQ dump 成 Hive，去進(jìn)行后續(xù)的離線(xiàn)數(shù)倉(cāng)的相關(guān)工作。切換成這套方案后，dump 操作就可以省掉，能夠做到流批的數(shù)據(jù)復(fù)用的能力來(lái)減少整體的中間存儲(chǔ)的成本。

第二個(gè)場(chǎng)景是多維分析場(chǎng)景，其特點(diǎn)是實(shí)時(shí)數(shù)據(jù)清洗后直接支持看板等實(shí)時(shí)的OLAP查詢(xún)。基于批流一體方案結(jié)合 Presto 查詢(xún)來(lái)滿(mǎn)足業(yè)務(wù)側(cè)分鐘級(jí)時(shí)延訴求，和秒級(jí)查詢(xún)響應(yīng)訴求。我們團(tuán)隊(duì)針對(duì)Presto進(jìn)行了許多優(yōu)化，包括native engine等相關(guān)技術(shù)，以實(shí)現(xiàn)高性能查詢(xún)。目前，該場(chǎng)景也在現(xiàn)場(chǎng)落地，并取得了不錯(cuò)的收益。另外，因?yàn)檎w的流批是數(shù)據(jù)表，存儲(chǔ)是統(tǒng)一的，所以不需要額外將其轉(zhuǎn)儲(chǔ)為Hive表，也不需要維護(hù)離線(xiàn)存儲(chǔ)的快照。

第三個(gè)場(chǎng)景就是批流復(fù)用的日志場(chǎng)景，一般大家直覺(jué)上會(huì)從 ODS 層切換做批流復(fù)用，字節(jié)內(nèi)部，在實(shí)時(shí)場(chǎng)景會(huì)先對(duì)接埋點(diǎn)數(shù)據(jù)， Flink 端去做清洗，落到實(shí)時(shí)的存儲(chǔ)里面，然后對(duì)接看板等下游。在離線(xiàn)數(shù)倉(cāng)上，也會(huì)將所有埋點(diǎn)信息存儲(chǔ)下來(lái)，根據(jù)具體的業(yè)務(wù)場(chǎng)景落成不同的 ODS 表，再去構(gòu)建離線(xiàn)數(shù)倉(cāng)的任務(wù)。

當(dāng)我們整體把存儲(chǔ)換成 LAS 這套方案之后，只需要維護(hù) ODS 層的數(shù)據(jù)，就能支持離線(xiàn)和實(shí)時(shí)兩個(gè)的場(chǎng)景去進(jìn)行分析。

最后是飛書(shū)數(shù)倉(cāng)的場(chǎng)景落地，整體鏈路比較清晰，分為實(shí)時(shí)和離線(xiàn)兩個(gè)鏈路，這里離線(xiàn)實(shí)時(shí)鏈路主要是去做一些人事信息變更之類(lèi)的業(yè)務(wù)。離線(xiàn)鏈路主要是對(duì)一些長(zhǎng)周期的問(wèn)題去做一些數(shù)據(jù)的修正，把這部分修正的數(shù)據(jù)回補(bǔ)到我們的實(shí)時(shí)鏈路里面，讓下游的看板數(shù)據(jù)變得更準(zhǔn)確。

在實(shí)時(shí)數(shù)據(jù)傳輸和離線(xiàn)數(shù)據(jù)處理兩個(gè)環(huán)節(jié)中，我們都采用了LAS之類(lèi)的存儲(chǔ)來(lái)替換底層存儲(chǔ)。離線(xiàn)數(shù)據(jù)處理要求數(shù)據(jù)的處理時(shí)間在10-15分鐘內(nèi)完成，因此用戶(hù)更慣用Spark處理離線(xiàn)數(shù)據(jù)處理環(huán)節(jié)。針對(duì)此，我們提供了基于Spark Thrift Server的解決方案，以減少離線(xiàn)數(shù)據(jù)處理中每個(gè)環(huán)節(jié)的資源申請(qǐng)開(kāi)銷(xiāo)，維持常駐資源，讓用戶(hù)運(yùn)行SQL來(lái)構(gòu)建離線(xiàn)數(shù)據(jù)處理模塊。

在實(shí)時(shí)數(shù)據(jù)傳輸環(huán)節(jié)中，用戶(hù)原本采用Kafka構(gòu)建基礎(chǔ)表并使用Hbase進(jìn)行維表構(gòu)建。因?yàn)镠base對(duì)聯(lián)合主鍵的支持并不友好，用戶(hù)每次在讀寫(xiě)時(shí)都需要去序列化和反序列化主鍵列，并將復(fù)合主鍵拼裝成單一主鍵，最后將其寫(xiě)入Hbase中，然后再?gòu)腍base中讀取。這樣的流程很復(fù)雜，且時(shí)間耗時(shí)較長(zhǎng)。

那么除了替換之外，就像之前提到的一樣，我們會(huì)替換掉MQ或Kafka這樣的組件，并且我們還用其他存儲(chǔ)替代了Hbase。我們的主要目標(biāo)是實(shí)現(xiàn)基于Flink的lookup join功能，并將小表直接加載到內(nèi)存中，以提高lookup join的性能。因此，在實(shí)時(shí)鏈路這一塊，我們已經(jīng)替換了所有的存儲(chǔ)組件。

四、未來(lái)規(guī)劃

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

五、Q&A環(huán)節(jié)

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

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

Q：LAS支持的組件索引和二級(jí)索引是什么樣的索引結(jié)構(gòu)？

A：實(shí)際上，這部分主要是社區(qū)原生實(shí)現(xiàn)，包括我們內(nèi)部的實(shí)現(xiàn)也大部分已經(jīng)貢獻(xiàn)到社區(qū)了。就組件而言，社區(qū)實(shí)現(xiàn)是基于哈希去進(jìn)行分桶，并同時(shí)記錄一些布隆過(guò)濾器（Bloom filter）。關(guān)于二級(jí)索引，社區(qū)目前正在進(jìn)行迭代，但并沒(méi)有完全合入。

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

A：會(huì)。時(shí)效性提升很多。因?yàn)榈谝粋€(gè)是我們寫(xiě)內(nèi)存，第二個(gè)是整個(gè)數(shù)據(jù)結(jié)構(gòu)會(huì)相對(duì) Hudi 來(lái)說(shuō)會(huì)比較輕量一些。

Q：BTS 與 Hudi分別適用的應(yīng)用場(chǎng)景。BTS 與Hudi的具體關(guān)系。

A：首先對(duì)于我們來(lái)說(shuō)，我們整體的這套方案叫LAS， LAS 底層是基于 Hudi 去做的實(shí)現(xiàn)，為了支持流批一體存儲(chǔ)，我們?cè)贖udi上加了一層內(nèi)存緩存層BTS，結(jié)合查詢(xún)引擎，一整套方案稱(chēng)之為 LAS。因?yàn)?BTS 是基于 Hudi 上架了一層，所以 BTS 整體的邏輯會(huì)跟 Hudi 強(qiáng)相關(guān)，它我兩者之間的交互主要就是內(nèi)存文件到持久化存儲(chǔ)中間的一些交互，其他的大部分的設(shè)計(jì)會(huì)沿用Hudi的部分的邏輯，比方說(shuō)我們會(huì)通過(guò)TSM, Table Service Management.去做一些比較優(yōu)化的服務(wù)，比方說(shuō) Hudi 的compaction， BTS 的 WAL 的clean，就這些操作。

Q：LAS怎么保證查詢(xún)數(shù)據(jù)的準(zhǔn)確性？

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