掃二維碼與項目經(jīng)理溝通
我們在微信上24小時期待你的聲音
解答本文疑問/技術(shù)咨詢/運營咨詢/技術(shù)建議/互聯(lián)網(wǎng)交流
不同的 Join 方式非常依賴于對 Doris 中數(shù)據(jù)劃分方式的透徹理解。因此先在這里列舉出必要的基礎(chǔ)知識。

首先,在 Doris 中數(shù)據(jù)都以表(Table)的形式進行邏輯上的描述。
在 Doris 的存儲引擎中,用戶數(shù)據(jù)被水平劃分為若干個數(shù)據(jù)分片(Tablet,也稱作數(shù)據(jù)分桶 Bucket)。每個 Tablet 包含若干數(shù)據(jù)行。各個 Tablet 之間的數(shù)據(jù)沒有交集,并且在物理上是獨立存儲的。
一個 Tablet 只屬于一個數(shù)據(jù)分區(qū)(Partition)。而一個 Partition 包含若干個 Tablet。因為 Tablet 在物理上是獨立存儲的,所以可以視為 Partition 在物理上也是獨立的。Tablet 是數(shù)據(jù)移動、復(fù)制等操作的最小物理存儲單元。
若干個 Partition 組成一個 Table。Partition 可以視為是邏輯上最小的管理單元。數(shù)據(jù)的導(dǎo)入與刪除,僅能針對一個 Partition 進行。
Doris 支持兩層的數(shù)據(jù)劃分。第一層是 Partition,支持 Range 和 List 的劃分方式。第二層是 Bucket(Tablet),僅支持 Hash 的劃分方式。也可以僅使用一層分區(qū)。使用一層分區(qū)時,只支持 Bucket 劃分。
下圖說明 Table、Partition、Bucket(Tablet) 的關(guān)系:
特別注意:
Doris 中的 Partition 和 Bucket 定義可能和某些其它數(shù)據(jù)庫系統(tǒng)的定義有一些差異,下面配以一個具體的建表語句為例來說明:
CREATE TABLE IF NOT EXISTS example_db.expamle_range_tbl
(
`user_id` LARGEINT NOT NULL COMMENT “用戶id”,
`date` DATE NOT NULL COMMENT “數(shù)據(jù)灌入日期時間”,
`timestamp` DATETIME NOT NULL COMMENT “數(shù)據(jù)灌入的時間戳”,
`city` VARCHAR(20) COMMENT “用戶所在城市”,
`age` SMALLINT COMMENT “用戶年齡”,
`sex` TINYINT COMMENT “用戶性別”,
`last_visit_date` DATETIME REPLACE DEFAULT “1970-01-01 00:00:00” COMMENT “用戶最后一次訪問時間”,
`cost` BIGINT SUM DEFAULT “0” COMMENT “用戶總消費”,
`max_dwell_time` INT MAX DEFAULT “0” COMMENT “用戶最大停留時間”,
`min_dwell_time` INT MIN DEFAULT “99999” COMMENT “用戶最小停留時間”
)
ENGINE=OLAP
AGGREGATE KEY(`user_id`, `date`, `timestamp`, `city`, `age`, `sex`)
PARTITION BY RANGE(`date`)
(
PARTITION `p201701` VALUES LESS THAN (“2017-02-01”),
PARTITION `p201702` VALUES LESS THAN (“2017-03-01”),
PARTITION `p201703` VALUES LESS THAN (“2017-04-01”)
)
DISTRIBUTED BY HASH(`user_id`) BUCKETS 16
PROPERTIES
(
“replication_num” = “3”
);
綠色高亮:Partition,此例中使用一個 date 字段進行分區(qū)
藍色高亮:Bucket,此例中使用 user_id 字段為作為分布列
創(chuàng)建分區(qū)時不可添加范圍重疊的分區(qū)
有兩種分區(qū)方式:
| 分區(qū)方式 | 一般用法 |
|---|---|
| Range | 通常按時間分區(qū),以方便地管理新舊數(shù)據(jù) |
| List | 支持的類型更豐富,分區(qū)值為枚舉值。只有當(dāng)數(shù)據(jù)為目標(biāo)分區(qū)枚舉值其中之一時,才可以命中分區(qū) |
作為分布式的 MPP 數(shù)據(jù)庫, 在 Join 的過程中是需要進行數(shù)據(jù)的 Shuffle。數(shù)據(jù)需要進行拆分調(diào)度,才能保證最終的 Join 結(jié)果是正確的。舉個簡單的例子,假設(shè)關(guān)系 S 和 R 進行Join,N 表示參與 Join 計算的節(jié)點的數(shù)量;T 則表示關(guān)系的 Tuple 數(shù)目。
目前 Doris 支持的 Join 方式有以上 4 種,這 4 種方式靈活度和適用性是從高到低的,對數(shù)據(jù)分布的要求越來越嚴(yán),但 Join 計算的性能則通過降低網(wǎng)絡(luò)開銷而越來越好。
Join 方式的選擇是 FE 生成分布式計劃階段會考慮的事項之一。在 FE 進行分布式計劃時,優(yōu)先選擇的順序為(總是會優(yōu)先選擇預(yù)期性能最好的):Colocate Join -> Bucket Shuffle Join -> Broadcast Join -> Shuffle Join。
Colocate 以及 Bucket Shuffle 是可遇不可求的。當(dāng)無法使用它們時,Doris會自動嘗試進行 Broadcast Join,如果預(yù)估小表過大則會自動切換至 Shuffle Join。
但是用戶可以通過顯式 Hint 來強制使用期望的 Join 類型,比如:
select * from test join [shuffle] baseall on test.k1 = baseall.k1;
原理比較簡單,這里不展開。
當(dāng) Join 條件命中了左表的數(shù)據(jù)分布列時,Broadcast 以及 Shuffle Join 會有非必要的網(wǎng)絡(luò)傳輸開銷。而 Bucket Shuffle Join 旨在解決這類問題,通過對左表實現(xiàn)本地性計算優(yōu)化,來減少左表數(shù)據(jù)在節(jié)點間的傳輸耗時,從而加速查詢。
以上的例子中,Join 的等值表達式命中了表 A(左表)的數(shù)據(jù)分布列。Bucket Shuffle Join 會根據(jù)表 A 的數(shù)據(jù)分布信息,將表 B(右表)的數(shù)據(jù)發(fā)送到對應(yīng)表 A 的數(shù)據(jù)計算節(jié)點。
定性分析上:
可以理解為在數(shù)據(jù)分布滿足一定條件的前提下,減少一切不必要的網(wǎng)絡(luò)傳輸開銷,實現(xiàn)完全的計算本地化來加速查詢。同時因為沒有網(wǎng)絡(luò)傳輸開銷,BE 節(jié)點可以擁有更高的并發(fā)度,從而進一步提升 Join 性能。
要理解這個算法,需要先了解兩個術(shù)語:
和 Buckets Sequence 這一概念:
一個表的數(shù)據(jù),最終會根據(jù)分桶列值 Hash、對桶數(shù)取模后落在某一個分桶內(nèi)。假設(shè)一個 Table 的分桶數(shù)為 8,則共有 [0, 1, 2, 3, 4, 5, 6, 7] 8 個分桶(Bucket),我們稱這樣一個序列為一個 BucketsSequence。每個 Bucket 內(nèi)會有一個或多個數(shù)據(jù)分片(Tablet)。當(dāng)表為單分區(qū)表時,一個 Bucket 內(nèi)僅有一個 Tablet。如果是多分區(qū)表,則會有多個(因為多個 Partition 中的不同 Tablet 會被劃分到相同的 Bucket)。
Colocation Join 功能,是將一組擁有相同 CGS 的 Table 組成一個 CG。并保證這些 Table 對應(yīng)的數(shù)據(jù)分片會落在同一個 BE 節(jié)點上。使得當(dāng) CG 內(nèi)的表進行分桶列上的 Join 操作時,可以通過直接進行本地數(shù)據(jù) Join,減少數(shù)據(jù)在節(jié)點間的傳輸耗時。
因此關(guān)鍵問題就轉(zhuǎn)變?yōu)榱恕溉绾伪WC這些 Table 對應(yīng)的數(shù)據(jù)分片會落在同一個 BE 節(jié)點上?」
通過同一 CG 內(nèi)的 Table 必須保證以下屬性相同實現(xiàn):
分桶列,即在建表語句中 DISTRIBUTED BY HASH(col1, col2, …) 中指定的列。分桶列決定了一張表的數(shù)據(jù)通過哪些列的值進行 Hash 劃分到不同的 Tablet 中。同一 CG 內(nèi)的 Table 必須保證分桶列的類型和數(shù)量完全一致,并且桶數(shù)一致,才能保證多張表的數(shù)據(jù)分片能夠一一對應(yīng)的進行分布控制。
同一個 CG 內(nèi)所有表的所有分區(qū)(Partition)的副本數(shù)必須一致。如果不一致,可能出現(xiàn)某一個 Tablet 的某一個副本,在同一個 BE 上沒有其他的表分片的副本對應(yīng)。不過,同一個 CG 內(nèi)的表,分區(qū)的個數(shù)、范圍以及分區(qū)列的類型不要求一致。
在固定了分桶列和分桶數(shù)后,同一個 CG 內(nèi)的表會擁有相同的 BucketsSequence。而副本數(shù)決定了每個分桶內(nèi)的 Tablet 的多個副本,存放在哪些 BE 上。假設(shè) BucketsSequence 為 [0, 1, 2, 3, 4, 5, 6, 7],BE 節(jié)點有 [A, B, C, D] 4個。則一個可能的數(shù)據(jù)分布如下:
+—+ +—+ +—+ +—+ +—+ +—+ +—+ +—+
| 0 | | 1 | | 2 | | 3 | | 4 | | 5 | | 6 | | 7 |
+—+ +—+ +—+ +—+ +—+ +—+ +—+ +—+
| A | | B | | C | | D | | A | | B | | C | | D |
| | | | | | | | | | | | | | | |
| B | | C | | D | | A | | B | | C | | D | | A |
| | | | | | | | | | | | | | | |
| C | | D | | A | | B | | C | | D | | A | | B |
+—+ +—+ +—+ +—+ +—+ +—+ +—+ +—+
CG 內(nèi)所有表的數(shù)據(jù)都會按照上面的規(guī)則進行統(tǒng)一分布,這樣就保證了,分桶列值相同的數(shù)據(jù)都在同一個 BE 節(jié)點上,可以進行本地數(shù)據(jù) Join。其核心思想是「兩次映射」,保證相同的 Distributed Key 的數(shù)據(jù)會被映射到相同的 Bucket Sequence,再保證 Bucket Sequence 對應(yīng)的 Bucket 映射到相同的 BE 節(jié)點:
通過查詢計劃可以檢查一個查詢是否使用了 Colocate Join,同時計劃中的 Exchange Node 也被去掉了,會將 ScanNode 直接設(shè)置為 Hash Join Node 的孩子節(jié)點。
DESC SELECT * FROM tbl1 INNER JOIN tbl2 ON (tbl1.k2 = tbl2.k2);
— 在 Hash Join 節(jié)點會顯示:
— colocate: true/false
Colocate Join 十分適合幾張表按照相同字段分桶,并高頻根據(jù)固定的字段 Join 的場景。這樣可以將數(shù)據(jù)預(yù)先存儲到相同的分桶中,實現(xiàn)本地計算。
Doris 在進行 Hash Join 計算時會在右表構(gòu)建一個 Hash Table,左表流式地通過右表的 Hash Table 從而得出 Join 結(jié)果。而 Runtime Filter 就是充分利用了右表的 Hash Table 構(gòu)建階段去做一些額外的事情。
在右表生成 Hash Table 的時,同時生成一個基于 Hash Table 數(shù)據(jù)的一個過濾條件,然后下推到左表的數(shù)據(jù)掃描節(jié)點。通過這樣的方式,Doris 可以在運行時進行數(shù)據(jù)過濾。
假如左表是一張大表,右表是一張小表,那么利用下推到左表的過濾條件就可以把絕大多數(shù) Join 層要過濾的數(shù)據(jù)在數(shù)據(jù)讀取時就提前過濾(如果能夠下推到引擎層,還能夠利用 Doris 針對 Key 列過濾的延遲物化),從而大幅度地提升 Join 查詢的性能。
Runtime Filter 在查詢規(guī)劃時生成,在 HashJoinNode 中構(gòu)建,在 ScanNode 中應(yīng)用。比如 T1(行數(shù) 10w) 和 T2(行數(shù) 2k) 的 Join 操作:
| > HashJoinNode <
| | |
| | 100000 | 2000
| | |
| OlapScanNode OlapScanNode
| ^ ^
| | 100000 | 2000
| T1 T2
|
顯而易見對 T2 掃描數(shù)據(jù)要遠遠快于 T1,如果我們主動等待一段時間再掃描 T1,等 T2 將掃描的數(shù)據(jù)記錄交給 HashJoinNode 后,HashJoinNode 根據(jù) T2 的數(shù)據(jù)計算出一個過濾條件,比如 T2 數(shù)據(jù)的最大和最小值,或者構(gòu)建一個 Bloom Filter,接著將這個過濾條件發(fā)給等待掃描 T1 的 ScanNode,后者應(yīng)用這個過濾條件,將過濾后的數(shù)據(jù)交給 HashJoinNode,從而減少 probe hash table 的次數(shù)和網(wǎng)絡(luò)開銷,這個過濾條件就是 Runtime Filter,效果如下:
| > HashJoinNode <
| | |
| | 6000 | 2000
| | |
| OlapScanNode OlapScanNode
| ^ ^
| | 100000 | 2000
| T1 T2
|
如果能將過濾條件(Runtime Filter)下推到存儲引擎,則某些情況可以利用索引(比如 Join 列為 Key 列,可以利用延遲物化能力)來直接減少掃描的數(shù)據(jù)量,從而大大減少掃描耗時,效果如下:
| > HashJoinNode <
| | |
| | 6000 | 2000
| | |
| OlapScanNode OlapScanNode
| ^ ^
| | 6000 | 2000
| T1 T2
|
可見,和謂詞下推、分區(qū)裁剪不同,Runtime Filter 是在運行時動態(tài)生成的過濾條件,即在查詢運行時解析 Join 條件確定過濾表達式,并將表達式下推給正在讀取左表的 ScanNode,從而減少掃描的數(shù)據(jù)量,進而減少 probe hash table 的次數(shù),避免不必要的 IO 和網(wǎng)絡(luò)傳輸。因為其運行時生效的特性,官方認為它是 Adaptive Query Execution 的一種應(yīng)用。
根據(jù)上面的例子,可以推導(dǎo)出場景滿足以下的條件時,使用 Runtime Filter 的效果會比較好:
Doris 支持 3 種 Runtime Filter:
工作原理和優(yōu)劣總結(jié)如下:
| Runtime Filter 類型 | 工作原理 | 適用場景 | 優(yōu)點 | 缺點 |
|---|---|---|---|---|
| IN | 子查詢的方式,實現(xiàn)上是將一個 Hashset 下推到 Scan 節(jié)點 | Broadcast Join | 開銷小,過濾效果明顯且快速 | 右表超過一定數(shù)據(jù)量時會失效,目前 Doris 配置的閾值是 1024 |
| Min/Max | 通過右表構(gòu)建一個 Range 范圍,然后將它下推到 Scan 節(jié)點 | 通用 | 開銷小 | 僅對數(shù)值類型有效果;對數(shù)值以外類型無法使用 |
| BloomFilter | 通過右表構(gòu)建一個 BloomFilter,然后將它下推到 Scan 節(jié)點 | 通用 | 通用性較好,適用于各種類型、效果也較好 | 配置比較復(fù)雜且計算成本較高;當(dāng)過濾率較低或者左表數(shù)據(jù)較少時,可能導(dǎo)致性能降低 |
一些使用的注意事項(比較細節(jié)了,后面考慮結(jié)合代碼再深入理解):
開啟 Runtime Filter 后,左表的 ScanNode 會為每一個分配給自己的 Runtime Filter 等待一段時間再掃描數(shù)據(jù),即如果 ScanNode 被分配了 3 個 Runtime Filter,那么它最多會等待 3000ms。
因為 Runtime Filter 的構(gòu)建和合并均需要時間,ScanNode 會嘗試將等待時間內(nèi)到達的 Runtime Filter 下推到存儲引擎,如果超過等待時間后,ScanNode 會使用已經(jīng)到達的 Runtime Filter 直接開始掃描數(shù)據(jù)。
如果 Runtime Filter 在 ScanNode 開始掃描之后到達,則 ScanNode 不會將該 Runtime Filter 下推到存儲引擎,而是對已經(jīng)從存儲引擎掃描上來的數(shù)據(jù),在 ScanNode 上基于該 Runtime Filter 使用表達式過濾,之前已經(jīng)掃描的數(shù)據(jù)則不會應(yīng)用該 Runtime Filter,這樣得到的中間數(shù)據(jù)規(guī)模會大于最優(yōu)解,但可以避免嚴(yán)重的劣化。
如果集群比較繁忙,并且集群上有許多資源密集型或長耗時的查詢,可以考慮增加等待時間,以避免復(fù)雜查詢錯過優(yōu)化機會。如果集群負載較輕,并且集群上有許多只需要幾秒的小查詢,可以考慮減少等待時間,以避免每個查詢增加 1s 的延遲。
有了前面兩表 Join 的 Runtime Filter 鋪墊,再來看 Join Reorder 的優(yōu)化,邏輯關(guān)系上就能夠理順了。
Doris 目前的 Join Reorder 算法是基于 RBO 的,邏輯描述如下:
可以發(fā)現(xiàn)前兩條,都是在朝著讓「右表」更小的方向去優(yōu)化,而最后一條則是從算法的性能上來考慮。
REF
#
#
以上就是Apache Doris Join 優(yōu)化原理詳解的詳細內(nèi)容,更多關(guān)于Apache Doris Join 優(yōu)化的資料請關(guān)注其它相關(guān)文章!
Linux 技術(shù)文檔 操作系統(tǒng)
數(shù)據(jù)庫運維技術(shù)服務(wù) ? Apache Doris Join 優(yōu)化原理詳解
分享到:
香港服務(wù)器選創(chuàng)新互聯(lián),2H2G首月10元開通。
創(chuàng)新互聯(lián)(www.cdcxhl.com)互聯(lián)網(wǎng)服務(wù)提供商,擁有超過10年的服務(wù)器租用、服務(wù)器托管、云服務(wù)器、虛擬主機、網(wǎng)站系統(tǒng)開發(fā)經(jīng)驗。專業(yè)提供云主機、虛擬主機、域名注冊、VPS主機、云服務(wù)器、香港云服務(wù)器、免備案服務(wù)器等。

我們在微信上24小時期待你的聲音
解答本文疑問/技術(shù)咨詢/運營咨詢/技術(shù)建議/互聯(lián)網(wǎng)交流