ByteHouse 是火山引擎上的一款云原生數(shù)據(jù)倉庫，為用戶帶來極速分析體驗(yàn)，能夠支撐實(shí)時(shí)數(shù)據(jù)分析和海量離線數(shù)據(jù)分析；便捷的彈性擴(kuò)縮容能力，極致的分析性能和豐富的企業(yè)級(jí)特性，助力客戶數(shù)字化轉(zhuǎn)型。

本文將從需求動(dòng)機(jī)、技術(shù)實(shí)現(xiàn)及實(shí)際應(yīng)用等角度，介紹基于不同架構(gòu)的 ByteHouse 實(shí)時(shí)導(dǎo)入技術(shù)演進(jìn)。

內(nèi)部業(yè)務(wù)的實(shí)時(shí)導(dǎo)入需求

ByteHouse 實(shí)時(shí)導(dǎo)入技術(shù)的演進(jìn)動(dòng)機(jī)，起初于字節(jié)跳動(dòng)內(nèi)部業(yè)務(wù)的需求。

在字節(jié)內(nèi)部，ByteHouse 主要還是以 Kafka 為實(shí)時(shí)導(dǎo)入的主要數(shù)據(jù)源（本文都以 Kafka 導(dǎo)入為例展開描述，下文不再贅述）。對(duì)于大部分內(nèi)部用戶而言，其數(shù)據(jù)體量偏大；所以用戶更看重?cái)?shù)據(jù)導(dǎo)入的性能、服務(wù)的穩(wěn)定性以及導(dǎo)入能力的可擴(kuò)展性。而對(duì)于數(shù)據(jù)延時(shí)性，大多數(shù)用戶只要是秒級(jí)可見就能滿足其需求?；谶@樣的場景，ByteHouse 進(jìn)行了定制性的優(yōu)化。

分布式架構(gòu)下的高可用

社區(qū)原生分布式架構(gòu)

ByteHouse 首先沿用了 Clickhouse 社區(qū)的分布式架構(gòu)，但分布式架構(gòu)有一些天然性架構(gòu)層面的缺陷，這些痛點(diǎn)主要表現(xiàn)在三個(gè)方面：

• 節(jié)點(diǎn)故障：當(dāng)集群機(jī)器數(shù)量到達(dá)一定規(guī)模以后，基本每周都需要人工處理節(jié)點(diǎn)故障。對(duì)于單副本集群在某些極端 case 下，節(jié)點(diǎn)故障甚至?xí)?dǎo)致數(shù)據(jù)丟失。
• 讀寫沖突：由于分布式架構(gòu)的讀寫耦合，當(dāng)集群負(fù)載達(dá)到一定程度以后，用戶查詢和實(shí)時(shí)導(dǎo)入就會(huì)出現(xiàn)資源沖突——尤其是 CPU 和 IO，導(dǎo)入就會(huì)受到影響，出現(xiàn)消費(fèi) lag。
• 擴(kuò)容成本：由于分布式架構(gòu)數(shù)據(jù)基本都是本地存儲(chǔ)，在擴(kuò)容以后，數(shù)據(jù)無法做 Reshuffle，新擴(kuò)容的機(jī)器幾乎沒有數(shù)據(jù)，而舊的機(jī)器上磁盤可能已經(jīng)快寫滿，造成集群負(fù)載不均的狀態(tài)，導(dǎo)致擴(kuò)容并不能起到有效的效果。

這些是分布式架構(gòu)天然的痛點(diǎn)，但是由于其天然的并發(fā)特性，以及本地磁盤數(shù)據(jù)讀寫的極致性能優(yōu)化，可以說有利有弊。

社區(qū)實(shí)時(shí)導(dǎo)入設(shè)計(jì)

• High-Level 消費(fèi)模式：依托 Kafka 自身的 rebalance 機(jī)制做消費(fèi)負(fù)載均衡。
• 兩級(jí)并發(fā)

基于分布式架構(gòu)的實(shí)時(shí)導(dǎo)入核心設(shè)計(jì)其實(shí)就是兩級(jí)并發(fā)：

一個(gè) CH 集群通常有多個(gè) Shard，每個(gè) Shard 都會(huì)并發(fā)做消費(fèi)導(dǎo)入，這就是第一級(jí) Shard 間的多進(jìn)程并發(fā)；

每個(gè) Shard 內(nèi)部還可以使用多個(gè)線程并發(fā)消費(fèi)，從而達(dá)到很高的性能吞吐。

• 攢批寫入

就單個(gè)線程來說，基本消費(fèi)模式是攢批寫入——消費(fèi)一定的數(shù)據(jù)量，或者一定時(shí)間之后，再一次性寫入。攢批寫入可以更好地實(shí)現(xiàn)性能優(yōu)化，查詢性能提升，并降低后臺(tái) Merge 線程的壓力。

無法滿足的需求

上述社區(qū)的設(shè)計(jì)與實(shí)現(xiàn)，還是無法滿足用戶的一些高級(jí)需求：

• 首先部分高級(jí)用戶對(duì)數(shù)據(jù)的分布有著比較嚴(yán)格的要求，比如他們對(duì)于一些特定的數(shù)據(jù)有特定的 Key，希望相同 key 的數(shù)據(jù)落盤到同一個(gè) Shard（比如唯一鍵需求）。這種情況下，社區(qū) High Level 的消費(fèi)模式是無法滿足的。
• 其次是 High level 的消費(fèi)形式 rebalance 不可控，可能最終會(huì)導(dǎo)致 Clickhouse 集群中導(dǎo)入的數(shù)據(jù)在各個(gè) Shard 之間分配不均。
• 當(dāng)然，消費(fèi)任務(wù)的分配不可知，在一些消費(fèi)異常情景下，想要排查問題也變得非常困難；對(duì)于一個(gè)企業(yè)級(jí)應(yīng)用，這是難以接受的。

自研分布式架構(gòu)消費(fèi)引擎 HaKafka

為了解決上述需求，ByteHouse 團(tuán)隊(duì)基于分布式架構(gòu)自研了一種消費(fèi)引擎——HaKafka。

高可用（Ha）

HaKafka 繼承了社區(qū)原有 Kafka 表引擎的消費(fèi)優(yōu)點(diǎn)，再重點(diǎn)做了高可用的 Ha 優(yōu)化。

就分布式架構(gòu)來談，其實(shí)每個(gè) Shard 內(nèi)可能都會(huì)有多個(gè)副本，在每個(gè)副本上都可以做 HaKafka 表的創(chuàng)建。但是 ByteHouse 只會(huì)通過 ZK 選一個(gè) Leader，讓 Leader 來真正地執(zhí)行消費(fèi)流程，其他節(jié)點(diǎn)位于 Stand by 狀態(tài)。當(dāng) Leader 節(jié)點(diǎn)不可用了，ZK 可以在秒級(jí)將 Leader 切到 Stand by 節(jié)點(diǎn)繼續(xù)消費(fèi)，從而實(shí)現(xiàn)一種高可用。

Low—Level 消費(fèi)模式

HaKafka 的消費(fèi)模式從 High Level 調(diào)整到了 Low Level 模式。Low Level 模式可以保證 Topic Partition 有序和均勻地分配到集群內(nèi)各個(gè) shard；與此同時(shí)，Shard 內(nèi)部可以再一次用多線程，讓每個(gè)線程來消費(fèi)不同 Partition。從而完全繼承了社區(qū) Kafka 表引擎兩級(jí)并發(fā)的優(yōu)點(diǎn)。

在 Low-Level 消費(fèi)模式下，上游用戶只要在寫入 Topic 的時(shí)候，保證沒有數(shù)據(jù)傾斜，那么通過 HaKafka 導(dǎo)入到 Clickhouse 里的數(shù)據(jù)肯定也是均勻分布在各個(gè) shard 的。

同時(shí)，對(duì)于有特殊數(shù)據(jù)分布需求——將相同 Key 的數(shù)據(jù)寫到相同 Shard——的高級(jí)用戶，只要在上游保證相同 Key 的數(shù)據(jù)寫入相同 Partition，那么導(dǎo)入 ByteHouse 也就能完全滿足用戶需求，很好地支持唯一鍵等場景。

場景一：

基于上圖可見，假設(shè)有一個(gè)雙副本的 Shard，每個(gè)副本都會(huì)有一張相同的 HaKafka 表處于 Ready 的狀態(tài)。但是只有通過 ZK 選主成功的 leader 節(jié)點(diǎn)上，HaKafka 才會(huì)執(zhí)行對(duì)應(yīng)的消費(fèi)流程。當(dāng)這個(gè) leader 節(jié)點(diǎn)宕機(jī)以后， 副本 Replica 2 會(huì)自動(dòng)再被選為一個(gè)新的 Leader，繼續(xù)消費(fèi)，從而保證高可用。

場景二：

在節(jié)點(diǎn)故障場景下，一般需要執(zhí)行替換節(jié)點(diǎn)流程。對(duì)于分布式節(jié)點(diǎn)替換有一個(gè)很繁重的操作——拷貝數(shù)據(jù)。

如果是一個(gè)多副本的集群，一個(gè)副本故障，另一個(gè)副本是完好的。我們很自然希望在節(jié)點(diǎn)替換階段，Kafka 消費(fèi)放在完好的副本 Replica 2 上，因?yàn)槠渖吓f數(shù)據(jù)是完備的。這樣 Replica 2 就始終是一個(gè)完備的數(shù)據(jù)集，可以正常對(duì)外提供服務(wù)。這一點(diǎn) HaKafka 是可以保證的。HaKafka 選主的時(shí)候，如果確定有某一個(gè)節(jié)點(diǎn)在替換節(jié)點(diǎn)流程當(dāng)中，會(huì)避免將其選為 Leader。

導(dǎo)入性能優(yōu)化：Memory Table

HaKafka 還做到了 Memory Table 的優(yōu)化。

考慮這樣一個(gè)場景：業(yè)務(wù)有一個(gè)大寬表，可能有上百列的字段 或者上千的 Map-Key。由于 ClickHouse 每一個(gè)列都會(huì)對(duì)應(yīng)落盤為一個(gè)具體的文件，列越多，每次導(dǎo)入寫的文件也就越多。那么，相同消費(fèi)時(shí)間內(nèi)，就會(huì)頻繁地寫很多的碎文件，對(duì)于機(jī)器的 IO 是很沉重的負(fù)擔(dān)，同時(shí)給 MERGE 帶來很大壓力；嚴(yán)重時(shí)甚至導(dǎo)致集群不可用。為了解決這種場景，我們?cè)O(shè)計(jì)了 Memory Table 實(shí)現(xiàn)導(dǎo)入性能優(yōu)化。

Memory Table 的做法就是每一次導(dǎo)入數(shù)據(jù)不直接刷盤，而是存在內(nèi)存中；當(dāng)數(shù)據(jù)達(dá)到一定量以后，再集中刷盤，減少 IO 操作。Memory Table 可以提供對(duì)外查詢服務(wù)的，查詢會(huì)路由到消費(fèi)節(jié)點(diǎn)所在的副本去讀 Memory Table 里邊的數(shù)據(jù)，這樣保證了不影響數(shù)據(jù)導(dǎo)入的延時(shí)性。從內(nèi)部使用經(jīng)驗(yàn)來看，Memory Table 不僅很好地解決了部分大寬表業(yè)務(wù)導(dǎo)入需求，而且導(dǎo)入性能最高可以提升 3 倍左右。

云原生新架構(gòu)

鑒于上文描述的分布式架構(gòu)的天然缺陷，ByteHouse 團(tuán)隊(duì)一直致力于對(duì)架構(gòu)進(jìn)行升級(jí)。我們選擇了業(yè)務(wù)主流的云原生架構(gòu)，新的架構(gòu)在 2021 年初開始服務(wù)字節(jié)內(nèi)部業(yè)務(wù)，并于 2023 年初進(jìn)行了代碼開源 [ByConity] https://github.com/ByConity/ByConity。

云原生架構(gòu)本身有著很天然的自動(dòng)容錯(cuò)能力以及輕量級(jí)的擴(kuò)縮容能力。同時(shí)，因?yàn)樗臄?shù)據(jù)是云存儲(chǔ)的，既實(shí)現(xiàn)了存儲(chǔ)計(jì)算分離，數(shù)據(jù)的安全性和穩(wěn)定性也得到了提高。當(dāng)然，云原生架構(gòu)也不是沒有缺點(diǎn)，將原來的本地讀寫改為遠(yuǎn)端讀寫，必然會(huì)帶來一定的讀寫性能損耗。但是，以一定的性能損耗來換取架構(gòu)的合理性，降低運(yùn)維成本，其實(shí)是利大于弊的。

上圖是 ByteHouse 云原生架構(gòu)的架構(gòu)圖，本文針對(duì)實(shí)時(shí)導(dǎo)入這塊介紹幾個(gè)重要的相關(guān)組件。

• Cloud Service

首先，總架構(gòu)分為三層，第一層是 Cloud Service，主要包含 Server 和 Catlog 兩個(gè)組件。這一層是服務(wù)入口，用戶的所有請(qǐng)求包括查詢導(dǎo)入都從 Server 進(jìn)入。Server 只對(duì)請(qǐng)求做預(yù)處理，不具體執(zhí)行；在 Catlog 查詢?cè)畔⒑?，把預(yù)處理的請(qǐng)求和元信息下發(fā)到 Virtual Warehouse 執(zhí)行。

• Virtual Warehouse

Virtual Warehouse 是執(zhí)行層。不同的業(yè)務(wù)，可以有獨(dú)立的 Virtual Warehouse，從而做到資源隔離?，F(xiàn)在 Virtual Warehouse 主要分為兩類，一類是 Default，一類是 Write，Default 主要做查詢，Write 做導(dǎo)入，實(shí)現(xiàn)讀寫分離。

• VFS

最底層是 VFS（數(shù)據(jù)存儲(chǔ)），支持 HDFS、S3、aws 等云存儲(chǔ)組件。

基于云原生架構(gòu)的實(shí)時(shí)導(dǎo)入設(shè)計(jì)

在云原生架構(gòu)下，Server 端不做具體的導(dǎo)入執(zhí)行，只做任務(wù)管理。因此在 Server 端，每個(gè)消費(fèi)表會(huì)有一個(gè) Manager，用來管理所有的消費(fèi)執(zhí)行任務(wù)，并將其調(diào)度到 Virtual Warehouse 上執(zhí)行。

因?yàn)槔^承了 HaKafka 的 Low Level 消費(fèi)模式，Manager 會(huì)根據(jù)配置的消費(fèi)任務(wù)數(shù)量，將 Topic Partition 均勻分配給各個(gè)任務(wù)；消費(fèi)任務(wù)的數(shù)量是可配置的，上限是 Topic Partition 數(shù)目。

基于上圖，大家可以看到左邊是 Manager ，從 catalog 拿到對(duì)應(yīng)的Offset，然后根據(jù)指定的消費(fèi)任務(wù)數(shù)目，來分配對(duì)應(yīng)的消費(fèi) Partition、并調(diào)度到 Virtual Warehouse 的不同節(jié)點(diǎn)來執(zhí)行。

新的消費(fèi)執(zhí)行流程

因?yàn)樵圃录軜?gòu)下是有事務(wù) Transaction 保證的，所有操作都希望在一個(gè)事務(wù)內(nèi)完成，也更加的合理化。

依托云原生新架構(gòu)下的 Transaction 實(shí)現(xiàn)，每個(gè)消費(fèi)任務(wù)的消費(fèi)流程主要包括以下步驟：

• 消費(fèi)開始前，Worker 端的任務(wù)會(huì)先通過 RPC 請(qǐng)求，向 Server 端請(qǐng)求創(chuàng)建一個(gè)事務(wù)
• 執(zhí)行 rdkafka::poll()，消費(fèi)一定時(shí)間（默認(rèn)8s）或者足夠大的 block
• 將 block 轉(zhuǎn)化為 Part 并 Dump 到 VFS（此時(shí)數(shù)據(jù)不可見）
• 通過 RPC 請(qǐng)求向 Server 發(fā)起事務(wù) Commit 請(qǐng)求

• （事務(wù)中 Commit 的數(shù)據(jù)包括：dump 完成的 part 元數(shù)據(jù)以及對(duì)應(yīng) Kafka offset）

• 事務(wù)提交成功（數(shù)據(jù)可見）

容錯(cuò)保證

從上述消費(fèi)流程里可以看到，云原生新架構(gòu)下的消費(fèi)，容錯(cuò)保證主要是基于 Manager 和 Task 的雙向心跳以及快速失敗策略：

• Manager 本身會(huì)有一個(gè)定期的探活，通過 RPC 檢查調(diào)度的 Task 是否在正常執(zhí)行；
• 同時(shí)每個(gè) Task 會(huì)在消費(fèi)中借助事務(wù) RPC 請(qǐng)求來校驗(yàn)自己的有效性，一旦校驗(yàn)失敗，它可以自動(dòng) kill；
• 而 Manager 一旦探活失敗，則會(huì)立即拉起一個(gè)新的消費(fèi)任務(wù)，實(shí)現(xiàn)秒級(jí)的容錯(cuò)保證。

消費(fèi)能力

關(guān)于消費(fèi)能力的話，上文提到它是一個(gè)可擴(kuò)展性的，消費(fèi)任務(wù)數(shù)量可以由用戶來配置，最高可以達(dá)到 Topic 的 Partition 數(shù)目。如果 Virtual Warehouse 中節(jié)點(diǎn)負(fù)載高的話，也可以很輕量地?cái)U(kuò)節(jié)點(diǎn)。

當(dāng)然，Manager 調(diào)度任務(wù)實(shí)現(xiàn)了基本的負(fù)載均衡保證——用 Resource Manager 來做任務(wù)的管理和調(diào)度。

語義增強(qiáng)：Exactly—Once

最后，云原生新架構(gòu)下的消費(fèi)語義也有一個(gè)增強(qiáng)——從分布書架構(gòu)的 At-Least-Once 升級(jí)到 Exactly—Once。

因?yàn)榉植际郊軜?gòu)是沒有事務(wù)的，只能做到一個(gè) At-Least-Once，就是任何情況下，保證不丟數(shù)據(jù)，但是一些極端情況可能會(huì)有重復(fù)消費(fèi)發(fā)生。到了云原生架構(gòu)，得益于 Transaction 的實(shí)現(xiàn)，每一次消費(fèi)都可以通過事務(wù)讓 Part 和 Offset 實(shí)現(xiàn)原子性提交，從而達(dá)到 Exactly—Once 的語義增強(qiáng)。

Memory buffer

對(duì)應(yīng) HaKafka 的 memory table，云原生架構(gòu)同樣實(shí)現(xiàn)了導(dǎo)入內(nèi)存緩存 Memory Buffer。

與 Memory Table 不同的是，Memory Buffer 不再綁定到 Kafka 的消費(fèi)任務(wù)上，而是實(shí)現(xiàn)為存儲(chǔ)表的一層緩存。這樣 Memory Buffer 就更具有通用性，不僅是 Kafka 導(dǎo)入可以使用，像 Flink 小批量導(dǎo)入的時(shí)候也可以使用。

同時(shí)，我們引入了一個(gè)新的組件 WAL 。數(shù)據(jù)導(dǎo)入的時(shí)候先寫 WAL，只要寫成功了，就可以認(rèn)為數(shù)據(jù)導(dǎo)入成功了——當(dāng)服務(wù)啟動(dòng)后，可以先從 WAL 恢復(fù)未刷盤的數(shù)據(jù)；之后再寫 Memory buffer，寫成功數(shù)據(jù)就可見了——因?yàn)?Memory Buffer 是可以由用戶來查詢的。Memory Buffer 的數(shù)據(jù)同樣定期刷盤，刷盤后即可從 WAL 中清除。

業(yè)務(wù)應(yīng)用及未來思考

最后簡單介紹實(shí)時(shí)導(dǎo)入在字節(jié)內(nèi)部的使用現(xiàn)狀，以及下一代實(shí)時(shí)導(dǎo)入技術(shù)的可能優(yōu)化方向。

ByteHouse 的實(shí)時(shí)導(dǎo)入技術(shù)是以 Kafka 為主，每天的數(shù)據(jù)吞吐是在 PB 級(jí)，導(dǎo)入的單個(gè)線程或者說單個(gè)消費(fèi)者吞吐的經(jīng)驗(yàn)值在 10-20MiB/s。（這里之所以強(qiáng)調(diào)是經(jīng)驗(yàn)值，因?yàn)檫@個(gè)值不是一個(gè)固定值，也不是一個(gè)峰值；消費(fèi)吞吐很大程度上取決于用戶表的復(fù)雜程度，隨著表列數(shù)增加，導(dǎo)入性能可能會(huì)顯著降低，無法使用一個(gè)準(zhǔn)確的計(jì)算公式。因此，這里的經(jīng)驗(yàn)值更多的是字節(jié)內(nèi)部大部分表的導(dǎo)入性能經(jīng)驗(yàn)值。）

除了 Kafka，字節(jié)內(nèi)部其實(shí)還支持一些其他數(shù)據(jù)源的實(shí)時(shí)導(dǎo)入，包括 RocketMQ、Pulsar、MySQL(MaterializedMySQL)、 Flink 直寫等。

關(guān)于下一代實(shí)時(shí)導(dǎo)入技術(shù)的簡單思考：

• 更通用的實(shí)時(shí)導(dǎo)入技術(shù)，能夠讓用戶支持更多的導(dǎo)入數(shù)據(jù)源。
• 數(shù)據(jù)可見延時(shí)和性能的一個(gè)折衷。