【轉】https://mp.weixin.qq.com/s/YNGdTbnG6iBjT1qtsK912w
Doris 簡介
首先簡單介紹一下 Doris 。Doris 是百度自主研發并開源的一個基于 MPP （大規模并行處理） 架構的分析型資料庫，它的特點就是性能卓越，能夠做到 PB 級别的資料分析的毫秒/秒級的響應，适用于高并發低延時下的實時報表、多元分析等需求場景。
Doris 最早是叫 Palo ，2017 年我們以百度 Palo 的方式在 GitHub 上進行了開源，在 2018 年的時候把它貢獻給 Apache 社群，正式更名為 Apache Doris 。而在百度内部一直沿用了 Palo 的名稱，并在百度智能雲上提供了 Palo 的企業級托管版本。
Doris 是一個 MPP 架構的分析型資料庫，有幾個特點：
第一個特點，簡單易用，支援标準 SQL 并且完全相容 MySQL 協定，産品使用起來非常友善。
第二，它采用了預聚合技術、向量化執行引擎，再加上列式存儲，是一個高效查詢引擎，能在秒級甚至毫秒級傳回海量資料下的查詢結果。
第三，它的架構非常簡單，隻有兩組程序：FE 負責管理中繼資料，并負責解析 SQL 、生成和排程查詢計劃；BE 負責存儲資料以及執行 FE 生成的查詢計劃。
這個簡潔高效的架構使得它運維、部署簡單、擴充性強，能夠支援大規模的計算。
通過下面這張圖我們簡單梳理一下 Doris 的結構，Doris 主要分為兩個角色，一個是 FE，另外一個是 BE 。
從 SQL 執行的角度說， FE 在 Doris 當中承擔了 MySQL 接入層，負責解析、生成、排程查詢計劃。BE 負責對應的查詢計劃的執行，負責實作實際的查詢、導入等工作。從資料的角度說，FE 負責中繼資料的存儲，比如表，資料庫，使用者資訊等資料，BE 負責列存資料的落地存儲。
這個架構是非常簡潔的，每個 BE 節點它是對等的。FE 分為 Leader、Follower、Observer這幾個角色，這和 ZooKeeper 之中的角色定位是類似的， Leader 跟 Follower 參與到叢集選主、中繼資料的修改等工作，而 Observer 是不參與這個過程的，隻提供資料的讀取，對外提供 FE 的讀擴充性，是以 FE 與 BE 節點都可以線性的擴充。
接下來是 Doris 當中資料的分布式存儲機制，Doris 作為一個 MPP 資料庫，它的資料存儲會深刻影響到後續我們要分析的 Join 實作與調優。
Doris 可以支援多副本的存儲，而且資料能夠自動遷移實作副本平衡。我們看到，Doris 中的資料是以 Tablet 的形式組織的，每一個表會拆分成多個 Tablet ，每個 Tablet 是由資料分區跟資料分桶來确定的。一旦确定了 Tablet 之後，在 Doris 當中所有的資料都是基于 Tablet 來排程，我們可以看到一個 tablet 可以分散在多個 BE 上做多副本的存儲，如果有 BE 節點當機，或者是有新的 BE 節點加入時，系統也會自動在背景執行資料副本的均衡。
在查詢的時候也會把查詢負載均衡到所有的 BE 上，這就是 Doris 在資料副本存儲上的整體架構，後面我們做 Join 分析的時候也會看到資料副本、包括資料是怎麼樣在當中排程的。
Doris Join 實作機制
Doris 支援兩種實體算子，一類是 Hash Join，另一類是 Nest Loop Join。
- Hash Join：在右表上根據等值 Join 列建立哈希表，左表流式的利用哈希表進行 Join 計算，它的限制是隻能适用于等值 Join。
- Nest Loop Join：通過兩個 for 循環，很直覺。然後它适用的場景就是不等值的 Join，例如：大于小于或者是需要求笛卡爾積的場景。它是一個通用的 Join 算子，但是性能表現差。
作為分布式的 MPP 資料庫， 在 Join 的過程中是需要進行資料的 Shuffle。資料需要進行拆分排程，才能保證最終的 Join 結果是正确的。舉個簡單的例子，假設關系S 和 R 進行Join，N 表示參與 Join 計算的節點的數量；T 則表示關系的 Tuple 數目。
Doris 支援 4 種資料 Shuffle 方式：
BroadCast Join
它要求把右表全量的資料都發送到左表上，即每一個參與 Join 的節點，它都擁有右表全量的資料，也就是 T(R)。
它适用的場景是比較通用的，同時能夠支援 Hash Join 和 Nest loop Join，它的網絡開銷 N * T(R)。
Shuffle Join
當進行 Hash Join 時候，可以通過 Join 列計算對應的 Hash 值，并進行 Hash 分桶。
它的網絡開銷則是：T（R） + T（N），但它隻能支援 Hash Join，因為它是根據 Join 的條件也去做計算分桶的。
Bucket Shuffle Join
Doris 的表資料本身是通過 Hash 計算分桶的，是以就可以利用表本身的分桶列的性質來進行 Join 資料的 Shuffle。假如兩張表需要做 Join，并且 Join 列是左表的分桶列，那麼左表的資料其實可以不用去移動右表通過左表的資料分桶發送資料就可以完成 Join 的計算。
它的網絡開銷則是：T（R）相當于隻 Shuffle 右表的資料就可以了。
Colocation
它與Bucket Shuffle Join相似，相當于在資料導入的時候，根據預設的 Join 列的場景已經做好了資料的 Shuffle。那麼實際查詢的時候就可以直接進行 Join 計算而不需要考慮資料的 Shuffle 問題了。（預shuffle）
Join 資料的 Shuffle 方式
下面這張圖是 BroadCast Join，左表的資料是沒有移動的，右表每一個 BE 節點掃描的資料都發送到對應的 Join 節點上，每個 Join 的計算節點上都有右表全量的資料。
第二種就是 Shuffle Join，每個資料掃描節點将資料掃出來之後進行Partition 分區，然後根據 Partition 分區的結果分别把左右表的資料發送到對應的 Join 計算節點上。
第三張圖是 Bucket Shuffle Join，右表資料掃描出來之後進行資料分區的 Hash 計算，根據左表本身的資料分布發送到對應的 Join 計算節點上。
最後就是 CoLocate Join。它其實沒有真正的資料 Shuffle，資料掃描之後進行 Join 計算就OK了。
Join 資料的 Shuffle 方式
上面這 4 種方式靈活度是從高到低的，它對這個資料分布的要求是越來越嚴格，但 Join 計算的性能也是越來越好的。
接下來就要分享的是 Doris 近期加入的一個新特性—— Runtime Filter 的實作邏輯。
Doris 在進行 Hash Join 計算時會在右表建構一個哈希表，左表流式的通過右表的哈希表進而得出 Join 結果。而 RuntimeFilter 就是充分利用了右表的 Hash 表，在右表生成哈希表的時，同時生成一個基于哈希表資料的一個過濾條件，然後下推到左表的資料掃描節點。通過這樣的方式，Doris 可以在運作時進行資料過濾。
假如左表是一張大表，右表是一張小表，那麼利用左表生成的過濾條件就可以把絕大多數在 Join 層要過濾的資料在資料讀取時就提前過濾，這樣就能大幅度的提升 Join 查詢的性能。
目前 Doris 支援三種類型 RuntimeFilter
- 一種是 IN—— IN，很好了解，将一個 hashset 下推到資料掃描節點。
- 第二種就是 BloomFilter，就是利用哈希表的資料構造一個 BloomFilter，然後把這個 BloomFilter 下推到查詢資料的掃描節點。。
- 最後一種就是 MinMax，就是個 Range 範圍，通過右表資料确定 Range 範圍之後，下推給資料掃描節點
RuntimeFilter 類型
Runtime Filter 适用的場景有兩個要求：
- 第一個要求就是左表大右表小，因為建構 Runtime Filter是需要承擔計算成本的，包括一些記憶體的開銷。
- 第二個要求就是左右表 Join 出來的結果很少，說明這個 Join 可以過濾掉左表的絕大部分資料。
當符合上面兩個條件的情況下，開啟 Runtime Filter 就能收獲比較好的效果。
當 Join 列為左表的 Key 列時，RuntimeFilter 會下推到存儲引擎。Doris 本身支援延遲物化，延遲物化簡單來說是這樣的：假如需要掃描 ABC 三列，在 A 列上有一個過濾條件: A 等于 2，要掃描 100 行的話，可以先把 A 列的 100 行掃描出來，再通過 A = 2 這個過濾條件過濾。之後通過過濾完成後的結果，再去讀取 BC 列，這樣就能極大的降低資料的讀取 IO。是以說 Runtime Filter 如果在 Key 列上生成，同時利用 Doris 本身的延遲物化來進一步提升查詢的性能。