分散式資料庫下子查詢和join等複雜sql如何實現？

01-01

之前也用過幾年的分散式資料庫或者叫做中間件吧，把一個表的數據按照一定的策略分散到各個資料庫節點上。但是隨之而來的問題也很顯而易見：
多節點數據查詢的複雜性，例如join和子查詢，導致目前這些貌似很難實現。
目前個人有個利用分步計算思想進行：

將輸入sql經過詞法，語法，語義分析，集合表結構信息和數據分布信息，生成包含多個階段(簡稱stage)的執行計劃

每個階段包括兩種sql，稱為mapsql和redsql，另外每個階段包括三個操作，map，數據洗牌和red；map和red分別執行mapsql和redsql；

每個階段的過程如下：

先在不同的資料庫節點中執行map操作，map操作執行mapsql，它的輸入是每個資料庫節點上的表裡面的數據，輸出根據某個欄位按照一定的規則進行分割，放到不同的結果集中，結果集作為數據洗牌的輸入

然後執行數據洗牌的過程，將不同結果集拷貝到不同的將要執行red的資料庫節點上

在不同的資料庫節點中執行red操作，red操作執行redsql；

最後返回結果

現舉例如下，歡迎拍磚。

例子1：
某一註冊時間範圍內的用戶信息 select * from tab_user_info t where u_reg_dt&>=? and u_reg_dt&<=? 生成的執行計劃一種可能為：在所有存儲節點上執行 select * from tab_user_info t where u_reg_dt&>=? and u_reg_dt&<=? 執行完成之後，根據u_id進行數據洗牌，這種情況下對洗牌沒有特殊的要求，直接放到目標計算節點上即可例子2：某一註冊時間範圍內的用戶的所有登錄信息 select t1.u_id,t1.u_name,t2.login_product from tab_user_info t1 join tab_login_info t2 on (t1.u_id=t2.u_id and t1.u_reg_dt&>=? and t1.u_reg_dt&<=?) 生成的執行計劃一種可能為：由於是join，所有的表都要進行查詢操作，並且為每張表打上自己的標籤，具體實施的時候可以加個表名字欄位，在所有存儲節點上執行 select u_id,u_name from tab_user_info t where u_reg_dt&>=? and t1.u_reg_dt&<=? select u_id, login_product from tab_login_info t 執行完成之後，這種情況下由於需要按照u_id進行數據洗牌，考慮到u_id的唯一值比較多，所以各個存儲節點上需要按照u_id進行劃分，例如有N個計算節點，那麼按照（最大u_id-最小u_id）/N平均劃分，將不同存儲節點上的同一範圍的u_id，劃分到同一個計算節點上然後在計算節點上執行如下操作 select t1.u_id,t1.u_name,t2.login_product from tab_user_info t1 join tab_login_info t2 on (t1.u_id=t2.u_id)

這是一個資料庫底層實現的問題，居然能夠引起280+的關注，國內做資料庫的幾路人馬都在這了。想起這幾年DRDS、OceanBase、TiDB等產品風起雲湧，好不熱鬧，不由感慨現在是做基礎技術的一個好年代，連資料庫研發這個圈子都能有這般人氣。 8年前，本人還在達夢做資料庫，舉目四望，除了同一個辦公室的幾個人外，滿世界找不到可以交流的同行。有一次接到一份來自遠方的，神秘的會議邀請郵件，上面寫著：「咱們都是做資料庫的...」，竟然有種地下黨接頭的感覺。

言歸正傳，談一談分散式資料庫 join 等複雜 sql 的問題。分散式資料庫下，複雜的sql當然是可以做的，關鍵是成本問題。從數據存儲的角度，分散式資料庫是把單機資料庫的那棵B樹給分散式化了（當然可能變成了LSM樹或者其他樹），假設有一個計算節點，它能知道參加join的每張表的結構，每張表數據的物理分布，而且這個節點的存儲能力無限大，那麼，在這個節點上，就能夠把所有join操作給做了，這種情況下，join演算法和單機資料庫的join演算法沒有區別。當然，任何計算節點的存儲能力，都不可能無限大，此時，就需要按照某種維度去切分、剪枝，然後一部分一部分地join，最後再將結果聚合，而這，就是所有分散式join方法的總體思路。

問題在於分散式關係型資料庫做join，和非關係型系統（Nosql、列存儲等）的相對成本。舉一個例子：請寫一條sql，統計出過去一年中，在淘寶上購買過服裝的女性用戶的平均年齡。對於這條查詢，基於嚴格的關係型模型（商品、訂單、用戶每種對象建一張表），無論怎麼優化查詢演算法，都比不上將三種對象合為一種對象，然後將過去一年的全部數據掃一遍快。

所以，關係模型不是萬能的，在單機資料庫的年代，很多問題沒有暴露，只要數據能在一塊，修修補補一下（比如建各種索引，比如違反嚴格的關係範式），好像問題都能夠解決，但到了分散式問題域下，關係模型就處處捉襟見肘了。而這，正是過去各種NoSql運動的母題。

現在的潮水，好像又回到關係模型的思路上來。但是，這不是復辟，而是有選擇地重建。不管是OceanBase還是TiDB，後續用在OLTP這種高吞吐，低延時的業務場景，會比較合適，而OLAP問題，則用hadoop、spark等效果已經被證明的技術會更合適。更值得思考的是，丟棄用過去二十年時間，用成千上萬用戶打磨出的，成熟穩定的mysql、pg不用，重新再造一個輪子，真的最好的選擇嗎？

另一種思路是，用多種數據存儲和處理技術，構建出一個數據池塘，來滿足企業各種業務場景下的數據存儲和處理需求，比如基於中間件+mysql來構建OLTP系統，基於hadoop、spark構建OLAP系統，採用redis來做緩存。。。這其實也是目前眾多公司正在做的。

必須得承認，這種做法需要不少技術投入和技術能力，不是所有公司都能玩得起。但假如在中間件這塊切入，消除運維的複雜性和風險程度，構建一個好用、好管理的數據池塘，是否就能夠將這種做法普及呢？如果將這個數據池塘搬到雲平台上，以在線服務的方式，提供給企業，那麼結果會怎樣？

UCloud上周剛內測的分散式資料庫產品：UDDB，是我們UDB團隊在這個方向上，做的一個小嘗試。UDDB現在還很小，功能也很簡單，只是基於中間件+mysql，打造了一個能夠對數據做水平切分的分散式資料庫，能夠做到不中斷業務的數據遷移和水平擴展，透明的讀寫分離，以及比開源中間件更好的SQL支持（對SQL的支持不斷迭代中）。最重要的是，它是運行在UCloud公有雲上的，具備UCloud產品一如既往的易用性和易管理性，和完善的技術支持能力。

基本上可以說，凡是你曾經試圖用資料庫中間件來解決的問題，UDDB都能夠解決，而且足夠好用，省心，成本足夠低。

如果有想參加內測的同學，可以知乎聯繫。

我們希望通過UDDB來驗證下我們的想法，如果受到用戶認可，我們將持續迭代，不為炫技，一切都圍繞為用戶創造價值。當然，目前的資料庫領域猶如三更半夜黎明之前，大家都在黑暗中摸索，期待用自己的產品點亮接來下的大數據時代。UDDB代表我們摸索的一個方向，希望和大家一起努力。

額，我說一下基於map reduce的分散式sql演算法吧……

1. 給入一個sql語句，經過sql planner做出plan，這個plan就是一個DAG圖，每條邊就是一個shuffle-sort的過程，一般aggr、window、join或者直接的sort、distribute by、寫partition這些操作會引入shuffle-sort。

題主在1中的例子壓根不需要shuffle-sort的過程，直接split一下表就可以搞了。

例2，需要再Plan階段做一下謂詞下推，把on條件推上去（當然外連接是不能隨便推的）。

2. join（等值連接）的分散式演算法主要分為hash-join和merge-sort join。 merge join要求左右表都是有序表，每次在兩邊同時讀相等的等值連接的key到內存中（其實在內存的只有一邊就可以了）做笛卡爾積。比如 a join b on a.key = b.key。a.key的值是1,2,3 b.key的值是1,3,4，那麼先a.key = 1和所有行和b.key = 1的所有行join，忽略掉a.key = 2，再讀key = 3。這樣就需要左父親的DAG節點（Task）按照a.key shuffle sort，右父親的DAG節點按照b.key shuffle-sort。

易見這種演算法理論上可以的無限的橫向拓展，只要保證每個instance上的key都是有序的，且同一個key都在一個instance上。

hash-join要求一個父親是小表，可以全部裝載到內存中，並組織成map這樣的數據結構。比如a join b其中a是小表，那麼每個instance都要有一份a的全量數據，b不要求是有序的，隨機分配到每個instance上即可，但是a表要廣播到每個Instance上。對於等值連接，b的每一行都可以在內存中查找具有相等key的全部a，做笛卡爾積即可。對於非等值連接，b的每一行和a的全量做笛卡爾積然後按照on條件過濾即可。

值得注意的是，a left join b ，a不能是小表，因為a不會知道自己有哪些行沒有被匹配到。

建議參考MSRA的madlinq和trinity相關論文，這種事情早已被艸過。

Oceanbase已做到，基本就是題主的思路，能下壓到各節點的做的盡量下壓，最後匯聚到一點規約，這個設計簡單，問題就是只能有一個點做規約。曾經還做過一版多輪MapReduce的方案，不過後來好像廢棄了。

更詳細的，坐等SQL組 @曉楚@茂七@楊志豐同學來回答

HBase不支持join

要靠上面搭一層Phoenix才行

默認的是left outer join

具體做法需要靠掛載的每台機器上的coprocessor

更多細節要看cod 機器間的message passing跑不了肯定不會滿足有locality (反正code我沒看逃…

join 主要演算法就那麼幾種。

nest loop

hash

merge sort 。

沒變過。分散式系統也如此。只是分散式系統上做join效率比單機低。要做一些對應的調優。另外因為數據量的變化原有的一些勉強能做的演算法變得不能做了。

基本就這些

沒做過，類似的文章倒是看過，有幾個相關問題要回答，

1. 一致性。很大的難題。可以選擇的是在snapshot上做放棄一致性，或者類似MegaStore有一個中心點做全局version管理，做分散式事務，或者類似spanner用絕對時間管理版本避免單點，或者根據業務情況定製一個簡化版本。

2. 性能。單機join訪問的是自己的磁碟和內存，且能做更好的cache策略，多台機器就會面臨網路交換數據的問題，分散式join演算法花很大的精力解決這個問題，核心思想是不去遠程讀，比如第一次讀之後就保存在本地，計算跟著存儲走之類；減少遠程讀次數和數據量，比如一次多讀一點，批量讀取甚至全表數據分發到全部相關節點，數據壓縮，遠程把結果計算好再向上構建（下推執行）；使用更好的硬體，比如更好的網卡，全內存全快閃記憶體等等；優化軟體，比如減少拷貝，繞過內核協議棧等等；也可以預計算，比如構建索引。

3. 擴展性，計算能力不足的時候能否通過簡單加節點解決？sql map的思路是因為mr的擴展能力強，那麼將sql變成mr執行之後就具備了mr的擴展能力，不過這是願望，能不能做到也得看系統怎麼設計的了

當然如何足夠健壯也是一個很大的話題。

這個問題可以分為兩個小問題： 1. 如何實現join, 2. 如何實現分散式join。

單機join 的實現是前提。第1個問題涉及SQL實現的很多核心問題，不能總是假定連接總是在兩個基表之間，可能還有相關子查詢，視圖，派生表等，因此join可能發生在查詢分析時無法確定大小的中間結果之間，這些都影響join的執行邏輯，比如是否選擇嵌套循環接還是 hash 連接等。

解決了第一個問題，則第二個分散式問題相對簡單。對於參與 join的分布數據集s1, s2, 按照數據集的估算大小，可以選擇以下的演算法之一：

1. 把小的數據集，比如s1, 都廣播到各節點，把數據補齊，然後再做join; 結果集仍然是分布的；繼續參與後續其他運算；

2. 選擇s1, 或s2, 按join key計算hash 值，把數據重新分發到各站點，再做join, 結果集仍然是分布的;

實際的分發還需要考慮分布列是否和連接KEY匹配或部分匹配，是否有索引可用，是否有分區，是否啟用了節點內並行能細節，以減少通訊代價，提升性能。

我們達夢資料庫對分散式集群支持比較成熟，幾乎支持單機SQL的所有特性，目前國內部署了很多生產系統，有不少是從Oracle RAC遷移過來，採用達夢資料庫後應用性能大幅提升。

目前的理解是: 做sql(子查詢，join等)有兩種方式，第一 MR, 是批處理的方式，中間結果落盤，spark好像有更高端的處理方式了。第二 MPP分散式資料庫，用傳統sql執行計劃，分散式環境下用Motiom.操作，也就是節點間重新分布數據方式，因為mpp的節點只能處理本地數據，要想複雜join必須將其他節點的數據通過執行計劃分發過來。可參考Greenplumdb得分散式執行計劃原理。

關鍵是執行計劃怎麼做，不光要考慮JOIN的問題，還要考慮先後的問題，在單機情況下資料庫管理所有執行計劃，在分庫分表的環境下如何將多個庫的統計信息集中起來讓你的具有多個JOIN的SQL變得有效？如果是NESTLOOP或者是HASHJOIN，以誰為起始點？與其這樣，不如在業務層面做，讓程序決定誰先誰後，但是這又牽涉到一個問題，數據是變化的，資料庫通過不停收集統計信息確保執行計劃的合理性，那麼程序上怎麼做？總之，難！

自己實現擴展的sql語法即可。。如果編程語言實現，做個linq即可。

1.1. 建立跨機器跨分表的分散式sql查詢引擎機制

沒有分散式的sql引擎當然也可以查詢分散式的資料庫架構，但是就得要用java等語言來自己實現where 等操作了。代碼量大，繁瑣。一旦涉及到goupby 等聚合函數，報表操作一般重度使用這類的聚合函數，就非常繁瑣以及難度較高了，對開發人員要求較高了。。

所以建立一個分散式sql引擎查詢機制也是必要的

1.1.1. Linq方案（使用java語言建造linq庫即可，實現跨機器分表的查詢，就像單表查詢）。語法可以仿照.net linq，也可以參考sql1.1.2. 擴展sql，使其可以支持跨機器查詢

這個難度稍微大點，要擴展sql關鍵字。自己實現對sql的解析，詞法分析，語法分析，語義分析與執行。。。

其實有個概念叫分散式資料庫,Oracle,Teradata,Netezza都有類似的產品,表的存儲也是分布到不同的節點上的.Teradata里叫AMP,一個SQL語句會被放到各個結點去運行,然後匯總回來.

具體可以看看http://www.teradatawiki.net/2013/09/partitioned-primary-index.html

我知道有很多支持SQL的產品是運行在Hadoop上的,我總覺得有點別捏.就好比坐上了未來汽車,就別去摸方向盤了. 特別是很多Key-Value的資料庫,本身沒有表的概念,非要去跑SQL,會讓人很別捏.

感覺這種是不是採用MR系較好，比如Spark和Hadoop上的分散式數據倉庫，這种放在資料庫中做的話，不會影響TP的性能嗎

占坑，讀過TDDL和ZDAL的源碼，原理簡單，實現遠比這個複雜。。。分散式場景下，沒有完美的SQL跨庫解決方案，與其解決這個問題，不如想想怎麼在業務上去做拆分

個人感覺不是那麼的複雜，首先sql語法解析這塊都很成熟了，sql到mapred的生成其實hive中就有現成的了，無非map/red就是sql。可能執行計劃的優化上面要費點功夫。拿到apache 或者大公司的話去實現應該不算困難的事情