小紅書基于數(shù)據(jù)湖的流批統(tǒng)一存儲實(shí)踐

作者：張億皓 2023-04-18 07:49:06

今天分享的主題是基于流存儲與數(shù)據(jù)湖打造小紅書流批統(tǒng)一存儲。

一、Lambda架構(gòu)與實(shí)時數(shù)倉開發(fā)痛點(diǎn)

1、小紅書的數(shù)據(jù)平臺概覽

首先來整體介紹一下小紅書的數(shù)據(jù)平臺。

首先在最底層是一個個 Cloud，包括計(jì)算、存儲等。在這一基礎(chǔ)之上，是數(shù)據(jù)采集層，采集一些原始數(shù)據(jù)，比如用戶行為日志數(shù)據(jù)、RDBMS 關(guān)系型數(shù)據(jù)庫的增量日志數(shù)據(jù)，以及其他一些文件系統(tǒng)等。

然后基于源頭數(shù)據(jù)層（ODS 層）之上是數(shù)據(jù)存儲和加工層，主要分為兩大塊：一是偏離線的部分，主要使用 Hive、Spark 計(jì)算，使用 AWS S3 存儲；二是偏實(shí)時的部分，主要使用 Flink 計(jì)算，使用 Kafka 存儲。

再往上是一個數(shù)據(jù)共享層，我們把一些聚合數(shù)據(jù)、Join 數(shù)據(jù)和寬表數(shù)據(jù)寫入數(shù)據(jù)共享的一些分析引擎中，比如 ClickHouse、StarRocks、TiDB、HBase 等等。這些都是作為數(shù)據(jù)共享層數(shù)據(jù)存儲的底座，以及計(jì)算分析引擎的一個入口。

最上面是應(yīng)用層，我們基于這一層做報表、即時查詢等，還會對數(shù)據(jù)做封裝，打造一些統(tǒng)一的數(shù)據(jù)產(chǎn)品。

2、典型的 Lambda 架構(gòu)在小紅書的實(shí)踐現(xiàn)狀

小紅書采用的是典型的 Lambda 架構(gòu)。實(shí)時鏈路主要使用 Flink 和 Kafka；離線鏈路主要使用 S3、Spark 和 Hive。Lambda 的特點(diǎn)就是兩條鏈路互相獨(dú)立建設(shè)，互不影響。

3、實(shí)時數(shù)倉開發(fā)痛點(diǎn)

Lambda 架構(gòu)的痛點(diǎn)可以總結(jié)為三個方面：

① 實(shí)時和離線數(shù)據(jù)不一致，造成數(shù)據(jù)不一致的原因主要有三點(diǎn)：計(jì)算引擎不一致，相同 SQL 定義也容易產(chǎn)生不同結(jié)果；作業(yè)不同，開發(fā)人員需要維護(hù)兩套代碼，技術(shù)門檻高；數(shù)據(jù) TTL 不同，Join 分析天然誤差。

② Kafka 缺乏數(shù)據(jù)檢索能力，對用戶來說 Kafka 更像一個黑盒。不管 Kafka 中數(shù)據(jù)存儲的是一些類似 protobuf 的數(shù)據(jù)還是 json 格式的數(shù)據(jù)，在做檢索的時候都非常困難。如果用戶想要根據(jù)某個條件去檢索數(shù)據(jù)，這個數(shù)據(jù)很難被查找。KSQL 產(chǎn)品更像是一個 streaming 的處理，更注重的是實(shí)時流處理能力，用來做離線大規(guī)模檢索并不適合。

③ 流存儲存數(shù)據(jù)有限，回溯效率低。這一點(diǎn)最大的原因是成本高，數(shù)據(jù)不能無限存。而且如果要去回溯讀，從歷史上去回追數(shù)據(jù)，它讀的性能也不及批量讀。

二、流批統(tǒng)一存儲架構(gòu)介紹

基于 Lambda 帶來的痛點(diǎn)，我們萌生了去開發(fā)一個流批存儲的產(chǎn)品的想法來解決 Lambda 的痛點(diǎn)。下面就來介紹一些設(shè)計(jì)細(xì)節(jié)。

1、流批統(tǒng)一存儲架構(gòu)介紹

如下是流批統(tǒng)一存儲的整體架構(gòu)：

我們的流批統(tǒng)一存儲叫 Morphing Server，對用戶提供的 API 還是跟 Kafka 完全兼容，都是使用流式的方式去寫入和消費(fèi)，這些接口都沒有變，所以用戶的使用方式不會有任何變化。

區(qū)別在于用戶寫入數(shù)據(jù)到 Kafka，Kafka 內(nèi)部會有一個線程，異步將數(shù)據(jù)同步到數(shù)據(jù)湖中。我們的數(shù)據(jù)湖是采用的 Iceberg，當(dāng)數(shù)據(jù)寫入到 Kafka 中，內(nèi)部線程會去抓取 Leader 數(shù)據(jù)，經(jīng)過一些 Schema 數(shù)據(jù)解析轉(zhuǎn)換為 Table Format 格式寫入到 Iceberg 中，這個過程是異步的，對用戶來說是無感的。

Kafka 的數(shù)據(jù)會被其他 Flink 作業(yè)消費(fèi)，消費(fèi)完之后可以寫到下一個 Kafka 中，在下一個 Kafka 依然是以異步的形式將數(shù)據(jù)落地到數(shù)據(jù)湖中。數(shù)據(jù)湖中的數(shù)據(jù)就可以提供批讀取和批存儲的能力。對于 Iceberg 中的數(shù)據(jù)如何去讀取的問題，我們會根據(jù)實(shí)際情況選取一些高性能的分析引擎，比如 StarRocks、小紅書自研的 RedCK 等來讀取離線數(shù)據(jù)。

2、產(chǎn)品能力

這里我們總結(jié)了 6 點(diǎn)流批統(tǒng)一存儲所提供的能力。

① 流批統(tǒng)一：同時提供流存儲和批存儲的讀寫能力，構(gòu)建多種應(yīng)用場景。

② 無感寫入：對外提供的寫入接口為原生 Kafka API，用戶無需關(guān)注落數(shù)據(jù)湖過程，自動異步寫湖。

③ Schema 解析：數(shù)據(jù)在落湖前會提前進(jìn)行 Schema 解析，以結(jié)構(gòu)化、半結(jié)構(gòu)化的 Table 形式提供查詢。

④ 高速分析：借助 StarRocks 引擎的強(qiáng)大湖上查詢能力，能夠提供向量化、CBO 等高速查詢能力。

⑤ Exactly-Once：流、批數(shù)據(jù)實(shí)現(xiàn) Exactly-Once 語義，數(shù)據(jù)一致性高。

⑥ 支持 Rollback：支持批數(shù)據(jù)的 Rollback 能力，在 Schema 變更不及時下，回溯修復(fù)數(shù)據(jù)。

接下去，我們介紹一下技術(shù)選項(xiàng)是如何去考量？關(guān)于技術(shù)選項(xiàng)分為兩個部分：自動落湖的過程如何選擇；對于數(shù)據(jù)湖中的數(shù)據(jù)如何選取合適的引擎去更加高效讀取

3、選型考量：Builtin Or Extension？

對于自動落湖過程我們考慮了兩種形式，Builtin（內(nèi)嵌）和 Extension（外掛插件），這兩種形式其實(shí)都是可以的。

（1）Builtin 形式?

在 Builtin 的形式下，我們看到只有一個獨(dú)立的進(jìn)程，在里面處理落日志之外，還會有一個異步的線程叫 Iceberg Syncer 去不斷拉取日志中的數(shù)據(jù)，然后寫入湖中，這種方式有優(yōu)勢也有劣勢。

優(yōu)勢如下：

① 產(chǎn)品形態(tài)完整，統(tǒng)一入口。

② 不需要額外維護(hù)外部組件。

③ 資源利用率高，共享進(jìn)程。

劣勢如下：

① 企業(yè)內(nèi)生成集群版本難以升級，在企業(yè)中有一些集群并沒有流批一體的功能，在升級中會非常困難。

② 進(jìn)程隔離性弱，如果在異步線程中產(chǎn)生 bug，可能影響 Kafka 正常的讀寫功能。

（2）Extension 形式?

針對 Builtin 形式的一些劣勢，我們當(dāng)初考慮了另外一種選項(xiàng) - Extension，這個方式相對更加直觀。

Extension 形式，也存在著一些優(yōu)勢和劣勢。

優(yōu)勢如下：

① 接入靈活，集群不需要升級，我們把 Kafka 落湖進(jìn)程摘取到 Kafka 進(jìn)程之外，是一個單獨(dú)的進(jìn)程，這是最大的一個好處。

② 流存儲可替換，并不局限于 Kafka，可以替換成其他引擎。

③ 進(jìn)程隔離。

劣勢如下：

① 運(yùn)維成本高，組件依賴過多，需要維護(hù)兩套組件。

② 產(chǎn)品體驗(yàn)稍差，整體性弱。

目前我們落地的是 Builtin 的方式，所以后面介紹的一些細(xì)節(jié)方案都是基于 Builtin 方式的。

4、查詢 & 分析引擎選擇

接下來介紹查詢分析引擎的選型。我們希望找到一款 OLAP 產(chǎn)品，具備以下特點(diǎn)：

① MPP 架構(gòu)、向量化和 CBO 來提高分析性能。

② 支持多場景，能夠在各種場景下滿足我們的需求。

③ 大規(guī)模，離線分析數(shù)量大，數(shù)據(jù)種類多的情況下，在大規(guī)模數(shù)據(jù)量下性能不退化。

基于這些考量，有兩大類選擇：左邊的是 Apache Doris 和 StarRocks 為代表的 OLAP 分析引擎；右邊是 ClickHouse 和小紅書基于 ClickHouse 自研的 RedCK 分析引擎。

左邊的分析引擎對分布式支持更好，對 SQL 協(xié)議兼容性高，提供更加一站式的查詢平臺。右邊的分析引擎對單表性能更加優(yōu)秀，在超大規(guī)模下的數(shù)據(jù)承載能力更強(qiáng)，特別是我們在 RedCK 上做了一些深度的定制化自研去滿足更多應(yīng)用場景。

（1）StarRocks（湖上分析）?

下面介紹我們在分布式引擎上選擇的 StarRocks。

StarRocks 支持湖上分析能力。它本身支持讀數(shù)據(jù)湖，不需要將數(shù)據(jù)以任何形式同步到 StarRocks 上，更像一種外表的形式，可以通過 Iceberg 的 Catalog 去查詢數(shù)據(jù)，還會做一些 Cache 緩存來加速查詢。

（2）StarRocks vs Persto 在流批一體（Iceberg）上的查詢對比?

我們對 StarRocks 和 Presto 在流批一體上做了查詢性能的對比，主要分為兩大類，四小類的 SQL 進(jìn)行比對。

左邊主要是 Scan 全表掃描相關(guān)，在這一方面 Presto 的性能更加優(yōu)越，但是兩者差距不大。右邊主要是 GroupBy 相關(guān)的聚合場景，具有 MPP 架構(gòu)的 StarRocks 在性能上明顯更加優(yōu)于 Presto。這也是我們選擇 StarRocks 的原因。因?yàn)樵谶@個應(yīng)用場景下 Join 使用較少，所以這里沒有進(jìn)行對比。

（3）RedCK 架構(gòu)?

還有一類分析引擎就是之前提到的 ClickHouse 和 RedCK，如何去更好的分析湖上的數(shù)據(jù)，這里介紹一下我們自研的 RedCK。

它是一個存算分離的架構(gòu)，主要分為三個模塊：Service、Query Processing 和 Storage。

Service 主要提供 Gateway 網(wǎng)關(guān)和 Service Discovery 服務(wù)發(fā)現(xiàn)，能夠讓業(yè)務(wù)更好的接入；Query Processing 是計(jì)算層，可以去解析 SQL 生成執(zhí)行計(jì)劃，分派這些任務(wù)去讀寫；Storage 是存儲層，支持文件存儲比如 HDFS 和 Juice FS，還支持對象存儲比如 OBS 和 COS。

（4）RedCK（湖上分析）?

接下來看一下 RedCK 和流批存儲是如何結(jié)合的。

RedCK 通過 MergeTree 的格式跟其他查詢引擎打通，比如 Spark、Flink 等計(jì)算可以直接讀寫 MergeTree 上的數(shù)據(jù)，然后通過 RedCK 在 MergeTree 上做 OLAP 分析。這樣的好處是使用 Spark 在寫數(shù)據(jù)的時候可以有一個更好的性能，做到了讀和寫兩種引擎的解耦。

基于這個考慮，我們在 Kafka 流批一體的引擎在落湖的過程中，原本只支持傳統(tǒng)的 Parquet 現(xiàn)在也支持寫 MergeTree 格式，同時也去提交一些和 RedCK 相兼容的元數(shù)據(jù)信息。這樣 RedCK 可以根據(jù)元數(shù)據(jù)信息直接找到 MergeTree 去做一些分析。

5、架構(gòu)設(shè)計(jì)細(xì)節(jié)

整體上，落湖分為兩大塊：Commit 模塊和 Broker 模塊。

Commit 模塊主要負(fù)責(zé)：

① Iceberg 的元信息的管理。

② 協(xié)調(diào) Broker 觸發(fā) Broker 做 Checkpoint。

③ 更新寫入 Iceberg 的 WaterMark 和 CheckpointID。

④ Controller 做 RollBack 工作。

Broker 模塊主要負(fù)責(zé)的是數(shù)據(jù)湖寫入，利用 Kafka 本身的 Fetch 機(jī)制，將 Leader 上的最新數(shù)據(jù)進(jìn)行解析并且不斷寫入，按照 Partition 維度來做單獨(dú)的線程寫入數(shù)據(jù)。

（1）Broker 設(shè)計(jì)細(xì)節(jié)?

Broker 的設(shè)計(jì)主要包括如下內(nèi)容：

① Replica Leader：Kafka 原生部分，處理 Produce 請求和 Consume 請求。

② ReplicaRemoteFetcherThread：主要工作線程，異步 Fetch Leader 數(shù)據(jù)，經(jīng)過 Schema 解析，寫入 Iceberg。

③ DefultSchemaTransform：Schema 解析模塊，提供寫入 Schema Server 變更。

④ IcebergRemoteLogStorageManager：封裝 Iceberg 接口，提供寫入 Iceberg 的 API 集合。

⑤ Schema Server：提供 Schema 管理服務(wù)，支持 Protobuf、Json 等。

（2）Commiter 設(shè)計(jì)細(xì)節(jié)?

Committer 主要的工作內(nèi)容包括：

① Controller：暫時復(fù)用 Kafka Controller，實(shí)現(xiàn) Commit 邏輯。

② 與 Broker 交互：發(fā)送 Checkpoint 請求，協(xié)調(diào)各 Broker Checkpoint 信息。

③ 與 Iceberg 交互：發(fā)起 Commit 請求。

（3）Excatly-Once 實(shí)現(xiàn)：兩階段提交?

Exactly-once 語義主要依托于兩階段提交來實(shí)現(xiàn)數(shù)據(jù)不丟不重，具體如下：

① 第一步，Committer 向所有 Broker 發(fā)起一個 RPC 請求，也就是 Checkpoint 請求。

② 第二步，Broker 在接受到 Broker 請求之后將目前為止還沒 Flush 的數(shù)據(jù) Flush 到 Iceberg，完成之后將 Checkpoint 信息記錄到 Checkpoint Storage 中。

③ 第三步，Broker 向 Commiter 返回一個 ACK，告訴 Commiter 已經(jīng)完成 Flush 工作。

④ 第四步，Commiter 等到所有 Broker 返回的 ACK 信息之后，發(fā)起第一階段提交并且記錄到 Checkpoint Storage 中，實(shí)際上做一個 Commiter 和 CheckpointID 關(guān)聯(lián)。

⑤ 最后一步，等第一階段完成之后，發(fā)起第二階段提交，發(fā)出一個 Commit 提交告訴 Iceberg 可以落盤。

（4）Exactly-Once 實(shí)現(xiàn)：故障 Failover?

實(shí)際生產(chǎn)中，常會出現(xiàn)一些故障。接下來介紹各種故障情況下，如何保證數(shù)據(jù)的不丟不重。

故障情況大概分為如下幾類：

① Broker 故障：比如突然宕機(jī)，其實(shí)這個故障沒有太大關(guān)系，因?yàn)?Kafka 本身有 Leader 切換能力，Leader 切換到其他 Broker 之后，會在新的 Broker 拉起異步線程寫 Iceberg。它會從 Checkpoint Storage 中讀取上一次 Checkpoint，從上一次的 Checkpoint 恢復(fù)這些數(shù)據(jù)去重新寫操作。在一次 Checkpoint 數(shù)據(jù)向 Iceberg 的數(shù)據(jù)，因?yàn)槭?committer 還沒有進(jìn)行第二階段提交，對于 Iceberg 來說是不可見的，可以直接丟棄這些不可見的數(shù)據(jù)。

② Controller 故障：在第一階段提交的時候失敗，會被自動切換到別的機(jī)器上面去再起一個 Commiter 線程，會發(fā)現(xiàn)第一階段還沒完成，那么會重新向所有 Broker 發(fā)起一輪新的 RPC 請求，重新做一次 Checkpoint，這一次其它 Broker 在接受到 RPC 請求之后會發(fā)現(xiàn)不需要做 flush 操作，就會立刻返回 ACK。在收到所有 ACK 之后，會重新做一次第一階段提交；第一階段提交之后成功了，但是在第二階段提交的時候失敗了，那么 Controller 切換到另外的一個機(jī)器首先會去 Checkpoint Storage 中查詢，如果第一階段提交信息已經(jīng)存在就會直接發(fā)起第二階段提交工作。

③ Object Store 故障/HMS 故障：我們會做一個無限重試，并且將一些告警信息發(fā)送出來。

三、流批統(tǒng)一存儲應(yīng)用實(shí)踐

流批統(tǒng)一存儲在公司內(nèi)部落地之后，可以解決一些 Lambda 架構(gòu)帶來的問題，下面將從四個方面來介紹。

1、Kafka 數(shù)據(jù)檢索

在流批一體之前，開發(fā)同學(xué)去檢索 Kafka 數(shù)據(jù)比較復(fù)雜，如左圖顯示：第一步需要去申請一個 topic，按照需要寫數(shù)倉作業(yè)；第二步找 DBA 申請一個 OLAP 表；第三步再去寫 Flink JOB 去解析 topic 數(shù)據(jù)寫到剛剛申請的 OLAP 表中，這個表純粹是用來查詢和排障，整個鏈路比較長。在使用流批一體之后，開發(fā)同學(xué)申請一個 Topic，然后往 Topic 中寫作業(yè)，這個時候開發(fā)同學(xué)可以直接查詢流批統(tǒng)一存儲。