作者 | 宇毅

前言

近年來，數(shù)據(jù)庫系統(tǒng)服務(wù)的數(shù)據(jù)量呈指數(shù)級增長，同時(shí)也面臨處理的業(yè)務(wù)需求愈發(fā)復(fù)雜、實(shí)時(shí)性要求越來越高等挑戰(zhàn)。單機(jī)數(shù)據(jù)庫系統(tǒng)已經(jīng)逐漸不能滿足現(xiàn)代的數(shù)據(jù)庫服務(wù)要求，因此分布式數(shù)據(jù)庫/數(shù)據(jù)倉庫得到了越來越廣泛地運(yùn)用。

在實(shí)時(shí)分析(OLAP)領(lǐng)域，分布式數(shù)據(jù)倉庫可以充分發(fā)揮系統(tǒng)的分布式特點(diǎn)，將復(fù)雜的OLAP任務(wù)分解下發(fā)到系統(tǒng)中的所有節(jié)點(diǎn)進(jìn)行計(jì)算提升分析性能;分布式數(shù)據(jù)倉庫也可以比較方便地對系統(tǒng)節(jié)點(diǎn)進(jìn)行擴(kuò)容，應(yīng)對用戶業(yè)務(wù)數(shù)據(jù)量增加的需求。但是分布式數(shù)據(jù)倉庫用戶無法避免的一個(gè)問題是：隨著數(shù)據(jù)倉庫集群規(guī)模增大，擴(kuò)容帶來的性價(jià)比愈發(fā)降低。

造成這種現(xiàn)象的一個(gè)原因是，表連接(Join)作為數(shù)據(jù)庫業(yè)務(wù)中最廣泛使用的算子之一，在分布式計(jì)算中依賴系統(tǒng)節(jié)點(diǎn)間的數(shù)據(jù)交互;當(dāng)分布式集群規(guī)模增大時(shí)，節(jié)點(diǎn)之間的數(shù)據(jù)交互代價(jià)會(huì)明顯增加，這種情況下非?？简?yàn)分布式系統(tǒng)的網(wǎng)絡(luò)處理能力，并依賴用戶的數(shù)據(jù)表設(shè)計(jì)和SQL編寫能力以緩解數(shù)據(jù)交互壓力。

針對這個(gè)問題，業(yè)界不同的分布式數(shù)據(jù)庫系統(tǒng)提出了不同的Join運(yùn)行時(shí)過濾(Runtime Filter)算法。AnalyticDB for PostgreSQL(以下簡稱ADB PG)是一款PB級的MPP架構(gòu)云原生數(shù)據(jù)倉庫，同樣也面臨著上述問題的挑戰(zhàn)。本文從ADB PG架構(gòu)設(shè)計(jì)的角度出發(fā)，探討Runtime Filter在ADB PG中的實(shí)現(xiàn)方案，并介紹了基于Bloom Filter的ADB PG Dynamic Join Filter功能技術(shù)細(xì)節(jié)。

ADB PG架構(gòu)簡介

ADB PG基于開源項(xiàng)目Greenplum構(gòu)建，在單機(jī)PostgreSQL的基礎(chǔ)上進(jìn)行擴(kuò)展，將多個(gè)PG服務(wù)同時(shí)啟動(dòng)在單個(gè)或多個(gè)服務(wù)器上并組成集群，以分布式的形式提供數(shù)據(jù)庫服務(wù)。ADB PG將每一個(gè)PG服務(wù)稱為一個(gè)Segment，并引入了Slice的概念。Slice用于解決分布式系統(tǒng)中的網(wǎng)絡(luò)結(jié)構(gòu)，當(dāng)數(shù)據(jù)庫涉及到MPP多階段計(jì)算時(shí)，例如Hash Join左右表的Join Key不滿足相同的Hash分布，那么就需要對Join Key通過網(wǎng)絡(luò)傳輸進(jìn)行重分布，ADB PG將網(wǎng)絡(luò)傳輸?shù)那昂箅A段切分為不同的Slices。以下是一個(gè)ADB PG集群示意圖。

在這種架構(gòu)下如何解決大規(guī)模集群下表連接Join的性能問題呢?業(yè)界解決這個(gè)問題的一個(gè)方案是引入網(wǎng)絡(luò)代理節(jié)點(diǎn)，同一機(jī)器內(nèi)的Segment將網(wǎng)絡(luò)數(shù)據(jù)發(fā)送至本地代理節(jié)點(diǎn)，由代理節(jié)點(diǎn)與其它機(jī)器上的代理節(jié)點(diǎn)進(jìn)行網(wǎng)絡(luò)收發(fā)工作以減少網(wǎng)絡(luò)擁塞。該方案對ADB PG架構(gòu)的挑戰(zhàn)較大，且沒有從根本上減少Join的網(wǎng)絡(luò)Shuffle開銷。因此為了從Join根源上減少Join計(jì)算的數(shù)據(jù)量，ADB PG設(shè)計(jì)并實(shí)現(xiàn)了Join Runtime Filter方案。

Runtime Filter和Bloom Filter

Runtime FIlter的目的是在Join計(jì)算前篩選掉一部分?jǐn)?shù)據(jù)，需要一個(gè)Filter的實(shí)現(xiàn)“載體”。在結(jié)合ADB PG的架構(gòu)設(shè)計(jì)、存儲(chǔ)層和網(wǎng)絡(luò)層的特點(diǎn)后，我們選擇使用Bloom Filter作為Runtime Filter的實(shí)現(xiàn)形式。

Bloom Filter是一種概率數(shù)據(jù)結(jié)構(gòu)，通常被用于判斷一個(gè)元素是否屬于一個(gè)集合。Bloom Filter的優(yōu)點(diǎn)是其空間效率非常高，計(jì)算性能通常也高;缺點(diǎn)是存在陽性誤判率false positive，但是不存在false negative，即Bloom Filter判斷一個(gè)元素是否屬于集合的結(jié)果不是單純的true or false，而是"possible true" or "false"。

上圖是一個(gè)標(biāo)準(zhǔn)Bloom Filter的計(jì)算思路示意圖，其中的0、1為Bloom Filter用于表示集合信息的bit array，即每一位用一個(gè)bit存儲(chǔ)。上方x,y,z表示向Bloom Filter中插入的三個(gè)元素，分別使用3種hash算法計(jì)算hash值后在bit array中置位。而下方為判斷元素w是否屬于集合，由于3個(gè)hash值中的某一位沒有在bit array中被置位，可以肯定的是w不屬于集合。

Bloom Filter通常由以下幾個(gè)參數(shù)描述：

• m --- Bloom Filter bit array的大小m bits
• k --- 使用的hash函數(shù)個(gè)數(shù)k
• p --- 誤判率
• n --- Bloom Filter插入的元素個(gè)數(shù)

我們省略推導(dǎo)過程，直接將各個(gè)參數(shù)的關(guān)系給出：

當(dāng)Bloom Filter足夠大時(shí)，可以簡化為：

在設(shè)計(jì)Bloom Filter時(shí)，n和m我們可以根據(jù)實(shí)際計(jì)算場景提前確定，上述公式可以視為自變量為k，應(yīng)變量為p的函數(shù)p(k)，此函數(shù)通常在k > 0時(shí)通常不是單調(diào)的(由n:m確定)。因此Bloom Filter在設(shè)計(jì)時(shí)要考慮如何確定hash函數(shù)k的個(gè)數(shù)以獲得最小的誤判率p。根據(jù)上式可以計(jì)算得到當(dāng)p為極小值時(shí)，對應(yīng)k的值為：

Bloom Filter的參數(shù)設(shè)計(jì)：

如何將Bloom Filter應(yīng)用至ADB PG Join過濾優(yōu)化，我們首先要設(shè)計(jì)選擇Bloom Filter的參數(shù)。對于Bloom Filter插入元素的個(gè)數(shù)n，可以直接使用執(zhí)行計(jì)劃中獲得的Join右表計(jì)劃行數(shù);而為了獲得理想的過濾率，減少誤判率p，ADB PG使用了PG高版本Bloom Filter的思路，設(shè)計(jì)Bloom FIlter大小Bytes為n的2倍，即總體n:m達(dá)到1:16。在這個(gè)設(shè)計(jì)下，可以計(jì)算得到最佳的k取值為11，p(k)函數(shù)如下圖所示，當(dāng)k = 11時(shí)可以取得最小的p = 0.046%

k = 11意味著對于每一個(gè)元素，都需要計(jì)算11個(gè)hash值以插入到Bloom Filter bit array中，這對于ADB PG是無法接受的，構(gòu)建Bloom Filter的代價(jià)明顯過大。在構(gòu)建Bloom Filter時(shí)，ADB PG會(huì)綜合誤判率、hash計(jì)算等因素考慮，選擇合適的k值。

在確定構(gòu)建Bloom Filter的基本原則后，接下來就是工程實(shí)現(xiàn)問題。Bloom Filter的工程實(shí)現(xiàn)非常簡單高效，通常我們可以直接使用bitset數(shù)組來建立Bloom Filter，通過位操作實(shí)現(xiàn)Bloom Filter的插入和查找。下圖為向一個(gè)Bloom Filter bitset數(shù)組中插入元素的計(jì)算示意圖。

Dynamic Join Filter in ADB PG

在完成ADB PG Hash Join的Bloom Filter設(shè)計(jì)后，接來下討論如何將Bloom Filter應(yīng)用至Join的Runtime Filter中。ADB PG將基于Bloom Filter的Runtime Filter命名為Dynamic Join Filter。

1.Dynamic Join Filter的實(shí)現(xiàn)方式

由于ADB PG優(yōu)化器通常會(huì)選擇將右表作為小表，左表作為大表，因此ADB PG將Dynamic Join Filter的設(shè)計(jì)特點(diǎn)為單向過濾的，即僅用于右表過濾左表，暫不考慮左表過濾右表的形式;同時(shí)我們也可以將Dynamic Join Filter靈活應(yīng)用于Hash Join左表鏈路不同算子的過濾中。

由于Hash Join的形式不同，Dynamic Join Filter的實(shí)現(xiàn)形式可以總結(jié)為Local Join和MPP Join兩種形式，并根據(jù)Runtime Filter是否具有下推算子的能力做進(jìn)一步區(qū)分。

Local Join

Local Join是指左右表的Join Key均滿足相同Hash分布，無需再Shuffle數(shù)據(jù)。此時(shí)Hash、Hash Join和左表Scan處于同一個(gè)Slice內(nèi)部，即同一個(gè)進(jìn)程中，我們可以直接在進(jìn)程空間內(nèi)將Bloom Filter傳遞給左表Scan算子過濾輸出。

MPP Join

MPP Join是指左右表的Join Key均不滿足相同Hash分布，需要針對Join Key Shuffle數(shù)據(jù)。在前文介紹過，ADB PG的Hash Join和Hash算子一定處于同一個(gè)Slice內(nèi)部，因此基于基本原則只需要考慮左表Shuffle的情況，即左表在Hash Join前存在Motion的場景。

MPP Join存在的另一種情況是，左表Motion下不是簡單的Scan，也沒有關(guān)聯(lián)信息將Join Key的Bloom Filter下推至Scan。那么以減少網(wǎng)絡(luò)傳輸數(shù)據(jù)量為最后準(zhǔn)則，將Bloom Filter過濾放在Motion前，減少M(fèi)otion Sender的數(shù)據(jù)。

2.Bloom Filter網(wǎng)絡(luò)傳輸

Dynamic Join Filter在各個(gè)計(jì)算節(jié)點(diǎn)上建立了一個(gè)Local Bloom Filter，每個(gè)計(jì)算節(jié)點(diǎn)需要收集所有其它節(jié)點(diǎn)的Bloom Filter，并在本地組成完整的Bloom Filter后才能開始過濾計(jì)算。我們將Bloom Filter的收發(fā)分為兩種模式：全量傳輸和位傳輸。在發(fā)送前我們可以判斷兩種模式的數(shù)據(jù)量大小，并自適應(yīng)選擇數(shù)據(jù)量小的模式。

Bloom Filter全量傳輸

Bloom Filter位傳輸

性能測試

接下來我們對ADB PG Dynamic Join Filter的性能表現(xiàn)測試。測試集群為ADB PG公有云搭建的實(shí)例，測試使用TPC-H 1TB測試集(scale = 10000)，測試通過開啟\關(guān)閉Dynamic Join Filter功能對比執(zhí)行性能。下圖展示了TPC-H執(zhí)行性能有差異的Query測試結(jié)果：

可以看到Dynamic Join Filter在Q5、Q8、Q9和Q17上均獲得了較大的性能提升，其中Q17的優(yōu)化性能最佳，執(zhí)行時(shí)間137s優(yōu)化至8s。而Q10存在略微的性能回退：10s回退至12s，原因在于Q10的Join Key是完全匹配的，Dynamic Join Filter無法做到動(dòng)態(tài)提前過濾，而優(yōu)化器未能準(zhǔn)確估算代價(jià)導(dǎo)致計(jì)劃仍然使用了Dynamic Join Filter。此外Q20也因?yàn)閮?yōu)化器下推規(guī)則的的原因沒有選擇Dynamic Join Filter，實(shí)際上經(jīng)過分析Q20與Q17類似，比較適合使用Dynamic Join Filter。為了解決這些問題，ADB PG優(yōu)化器相關(guān)功能仍在開發(fā)迭代中。

總結(jié)&未來規(guī)劃

Dynamic Join Filter根據(jù)ADB PG架構(gòu)設(shè)計(jì)、存儲(chǔ)層和網(wǎng)絡(luò)層特點(diǎn)，使用Bloom Filter作為Join Runtime Filter的實(shí)現(xiàn)形式，在TPC-H測試中取得了明顯的性能提升成果。未來我們將從以下幾個(gè)方面做進(jìn)一步的開發(fā)和優(yōu)化，提升客戶使用體驗(yàn)：

完善Dynamic Join Filter功能，支持各種模式的Hash Join，并進(jìn)一步推廣到Merge Sort Join、NestedLoop Join的優(yōu)化中;

提升優(yōu)化器的代價(jià)估算模型精度，完善優(yōu)化器下推規(guī)則;

Runtime Filter自適應(yīng)調(diào)度。

歡迎訪問云原生數(shù)據(jù)倉庫ADB PG主頁，了解更多：https://help.aliyun.com/product/35364.html