1 UData-解決資料使用的最后一公里

1.1 背景

在大資料的范疇，我們經歷了資料產業化的歷程，從各個生產系統將資料收集起來，經過實時和離線的資料處理最侄訓集在一起，成為我們的主題域資料，下一步挖掘資料的價值將成為關鍵，

資料應用直接體現資料的價值，資料應用多種多樣，它們使用資料的方式也各不相同，UData作為資料資產和資料應用之間的橋梁，它的第一目標是解決所謂的資料使用的最后一公里問題，

UData平臺以資料指標為基本的管理單位，通過四個階段對于資料使用提供支持，一體化整合資料鏈路的整個生命周期，接資料、管資料、找資料、用資料，

UData核心聚焦資料應用場景，從資料應用倒推打通資料接入、資料管理、資料查詢等環節，各種資料應用對于資料的使用方式，大部分分為兩個場景：

1. 應用在線及時訪問資料，大多數以介面的形式，UData平臺相對應的提供了資料服務的模塊；
2. 業務人員通過在線查找自己需要的資料指標（資料指標地圖），可視化的進行人工資料分析和展示，UData平臺同時提供了資料分析的模塊，

1.2 UData功能架構圖

上圖，UData功能架構自底部向上，包含了資料流轉使用的整個程序，平臺內的功能模塊從資料使用的流程角度，完整的涵蓋了資料使用最后一公里的整個生命周期，

1.3 Udata的資料管理

UData對于資料的使用，從物理和邏輯兩個層面進行了劃分，并且對于多個租戶同樣進行了資源和計算的隔離，

1.4 Udata目前能做什么？

1.4.1 指標配置化開發管理

1. UData資料接入可以將外部資料實時或者定時的匯入平臺，同時平臺提供了多種資料源的聯邦查詢；
2. 在線可視化的創建資料指標，并對資料指標進行打標簽；
3. 資料指標地圖使業務人員方便的查找自己需要的業務指標；
4. 資料指標的開發，管理，使用，幾個階段相互分離，職責劃分更加松耦合，業務注意力更加聚焦；

1.4.2 指標積木式編排和介面服務

1. UData從底層資料源開始至最上層封裝成為資料指標對外提供資料服務；
2. 資料指標在UData中可以像積木一樣通過可視化的方式進行任意組合；
3. UData提供了介面編排能力，可以在指標組合基礎之上，實作帶有業務邏輯的分支條件判斷；

1.4.3 指標及明細互動式關聯分析和協同分享

1. UData可以重用資料視圖和資料指標，創建資料集，以此為基礎向上進行資料分析；
2. 資料集的配置支持SQL模式和可視化配置模式，分別針對不同SQL水平的分析人員；
3. 面向資料分析應用，以應用場景為單位進行資料和計算函式的管理和組織，場景可共享；
4. 資料在線化實時分析，無需線上匯出資料；
5. 在線Excel操作，持久化Excel模式，資料實時重繪，Excel報表在線共享;

2 Udata-查詢引擎執行介紹-一條SQL的旅行

2.1 引擎架構

Udata查詢引擎基于StarRocks進行了部分改造，由兩部分組成FrontEnd(FE),BackEnd(BE)組成，

• FE：負責接收和回傳客戶端的請求，元資料和集群的管理，查詢計劃的生成和優化，協調BE進行查詢，
• BE: 主要負責SR表的資料存盤和查詢，外部表形式連接三方存盤，并執行查詢計劃中的具體節點，例如scan, 投影，聚合等，

執行主流程：

1. FE收到Sql客戶端發起的查詢請求，決議sql并制定查詢計劃；
2. FE下發執行計劃到BE, 并指定一個BE為Coordinator;
3. 各BE按照查詢計劃中的PlanFragment為執行單位，接收作業，完成作業，并將結果匯聚到Coordinator節點；
4. Coordinator的BE節點將資料回傳給FE;
5. FE向Sql客戶端回傳結果；

2.2 從SQL陳述句到執行的程序

2.2.1 程序概覽

用戶通過Mysql客戶端工具或者JDBC等方式，將需要執行的SQL陳述句進行輸入，輸入后的SQL陳述句經過語法決議，Binder,Transformer,Optimizer等程序，從基礎的sql陳述句，經過語法樹，Relation，邏輯計劃，分布式物理計劃等程序，最終在FE端通過Coordinator發送到BE側進行執行，并后續收集BE回傳的資料，回傳給呼叫客戶端，

2.2.2 舉例介紹

表結構：

desc remote_mysql_decimal;

SQL:

select count(`decimal`) as sum,`key` from remote_mysql_decimal where id <= 1000 group by `key` order by sum desc limit 10;

2.2.3 執行程序詳解

1.決議SQL陳述句

在這一步驟中，SQL陳述句會進行語法檢查，不符合規范的陳述句回傳錯誤，之后經過語法決議，會生成一個抽象語法樹，上面實體中的SQL陳述句（陳述句中有聚合，排序，謂詞條件，limit等元素）生成的語法樹結構如下：

2.系結資料表元資料資訊-生成Relation

生成語法樹之后，只是單純的SQL語法資訊，在SR中FE有一個重要的作用，就是保存資料表的元資料資訊（庫名，表名，列名，資料型別，對應的外表）等，
在這一步驟中，會將抽象語法樹和FE中的元資料資訊（Catalog）進行關聯，豐富SQL相關的資訊，將抽象語法樹生成Relation這種資料結構，

3.Transformer - 基于RBO,進行Rewrite生成邏輯執行計劃

從Relation到邏輯計劃，只是基于一些SQL改寫規則，將樹中的一些節點轉變會邏輯計劃節點，

如：

• FromClause 會轉換為邏輯計劃中的LogicalScanOperator這種掃表操作；
• WhereClause 會轉換成邏輯計劃中的LOGICAL_FILTER，指導后續進行進行條件過濾；
• OrderByElements 會轉換成邏輯計劃的LOGICAL_TOPN，指導后續進行排序和limit;
• SelectList 會轉換為邏輯計劃的LogicalProjectOperator，指導后續進行投影操作，減少網路資料傳輸;

本實體中的SQL會生成如下的邏輯計劃：

4.Optimizer - 基于CBO優化

在這一步驟中，會根據上一步生成的邏輯計劃，同時結合FE中保存的元資料資訊，基于CBO優化執行計劃，進行謂詞下推，Join order 調整等，
本實體中生成的Optimizer Plan如下：

5.分布式物理計劃的生成- BE執行的并行單位（PlanFragment）

BE是分布式的，查詢實際執行的時候，會將計劃分配給具體的BE，BE之間，BE和FE之間通過RPC通信傳輸資料，BE執行的最小并行單位是Fragment, 在這一步驟中會生成分布式的物理計劃，
本實體SQL生成的分布式物理計劃如下：

2.3 資料的輸出

2.3.1 PlanFragment在BE側的映射

物理執行計劃切分成PlanFragment之后，會發送到BE側執行，BE會根據Fragment中的樹形結構，生成對應的Node，完成各自的算子邏輯，算子之間通過不停的呼叫下層算子的get_next()函式，將資料用chunk的形式進行組織并流動起來，chunk的資料結構是一種列式的批結構，非常有利于向量化的執行，

2.3.2 執行模型

1.火山模型/迭代模型 （ Volcano Model ）

在這種模型中，每一種操作會抽象成一個Operator, 在執行側作為一個運算元，從頂到下呼叫next()介面，資料從底部的scan節點向上傳輸，但是每次只傳輸計算一條資料，也叫做（Tuple-at-a-time）,是一種拉取執行模式，
優點：每個Operator可以單獨實作邏輯，比較單間，靈活，
缺點：每次傳輸計算一條資料，導致next()函式呼叫次數過多，cpu效率低，

2.物化模式/Materialization Model

這種模型的處理方式，仍然是呼叫自頂向下，資料從底向上，但是每一個操作Operator一次性處理所有的輸入，處理完成之后，將結果一次性向上輸出，
此模式對于資料量較大的OLAP不太適合，但是比較適合資料量較小的OLTP系統，

3.向量化模式/批處理模型 （ Vectorized / Batch Model ）

這種模型和火山模型非常類似，不同之處是每個Operator的next()函式，會回傳一批的tuples資料，相當于是一種批處理的模型，這是一種上面兩種模型的折中方式，
SR的向量化執行器主要集中在算子向量化，運算式向量化，存盤向量化；充分利用SIMD指令優化，CPU Cache友好，

3 Udata查詢引擎-聯邦查詢的增強

3.1 Udata查詢引擎發展的三個階段

3.1.1 社區版FE + 自研JAVA版BE

Udata查詢引擎的第一階段，是參照StarRocks的C++版本BE實作了一個JAVA版本的BE,主要完成了Udata在第一個階段的進行聯邦查詢的資料服務的任務，并且在第一個版本基礎上，已經實作了聚合計算的下推，同時也經過了618的考驗，在執行引擎層面積累了大量的經驗，為我們開展引擎改造的第二階段提供了支持；

3.1.2 原生StarRocks + Udata改進

鑒于StarRocks表的優異性能，我們將查詢引擎切換回原生的SR, 同時將之前的積累的優化經驗，在原生SR上進行了實作，包括聚合查詢和Sort排序的下推，額外支持了外表資料源CK,Jsf,Http，進行了查詢函式format等的豐富，

3.1.3 未來探索方向

在下一個階段，Udata查詢引擎將會在SR的基礎之上，密切地配合社區，引入新版本的功能，同時進行資料湖的使用探索和高性能的點查實踐，以及跨SR集群的聯邦查詢等，

3.2 計算下推 - 極限壓榨底層引擎的計算能力

3.2.1 優化背景

StarRocks在聯邦查詢方面針對MySQL, ElasticSearch已經有了非常快的性能，StarRocks在聯邦查詢方面的設計思想是針對不同的查詢外部資料源，設計不同的Scan節點，并且盡可能的將謂詞下推到Scan節點，在Scan節點查詢到資料之后，上層會共用Project節點，Agg節點，TopN等這些節點的算子，基本的查詢架構類似下圖，

這種設計使StarRocks有非常好的擴展性，可以很容易的擴展到新一種的資料源，也正是這種高度可擴展的設計使我們有機會在聯邦查詢的細節層面，做進一步的優化，比如將一些算子的計算也盡可能的推到外部表引擎，可以節省一部分網路傳輸的時間，同時最大程度的壓榨底層引擎原生計算能力，通過我們的測驗這種計算下推也達到了數倍于原來的性能，

3.2.2 優化范圍

在優化之前我們針對底層引擎和算子的特征做了調研，優化的范圍包括如下：

• 針對ES引擎，進行了聚合算子的下推，但是某些特殊算子排除，不支持sum(distinct ), avg(distinct ) 算子下推；
• 針對MySQL引擎和ClickHouse，進行了聚合算子，TopN算子的下推；
• 針對新增加的Jsf和Http，進行了查詢引數下推，運行時列過濾；

3.2.3 整體優化思路

目前整體的優化思路，主要分為兩個部分，FE側的改造和BE側的擴充，同時對于原生StarRocks計算方式保持兼容，可以輕易的切換回原來的計算模式，

1）FE 側改造優化- Optimizer Plan 的轉換

執行計劃優化流程

目前Udata查詢引擎對執行計劃進行優化的節點是在原來的Optimizer之后，我們從Scan節點開始對于執行計劃，進行了模式匹配，命中模式之后，進行對應的計算下推和投影的合并，同時過濾底層引擎不支持的特殊算子（ 如ES的sum(distinct) ），最終將轉變后的物理計劃發送給BE側進行執行，

模式匹配和計劃改寫

物理計劃的樹狀封裝：

ElasticSearch:

Mysql:

查詢樹改寫：

最終，AggScanOperator 會轉變為AggNode,發送到BE進行執行，

2）BE 側改造優化

針對執行計劃進行了改寫之后，同樣在BE側我們創建了對應的Node節點，完成計算下推后的執行邏輯，向下對接外部執行引擎，同時向上對接類似join的聚合節點，最終輸出結果資料，

3）原生SR兼容

同時執行層面，我們設定了靈活的開關 （ set agg_push_down = 0 ），可以非常容易的關閉UData優化，

3.2.4 改造成效-（ 30秒 vs 6秒 ）

在我們的實際程序中，我們對于計算下推，尤其是多表聚合后關聯的場景進行了觀察測驗，計算性能隨著聚合表數目的增加，會有成倍數的效果提升，

3.3 JSF&HTTP&ClickHouse的支持 - 京東生態的對齊

3.3.1 簡介

JSF是京東內部的一種RPC呼叫服務，很多資料分析的場景中，一些維表是在其他服務中用JSF或者Http的方式提供的，或者一些已經計算好的資料指標需要在我們的UData計算引擎中進行關聯查詢，因此我們增加了對于JSF和Http的支持，來作為京東生態的一個補充，

JSF和HTTP查詢的兩個關注點是如何將查詢引數進行下推和如何將回傳的結構化資料映射為表中的列資料，以便在聯邦查詢中進行資料關聯和聚合，

同時，京東內部有不少使用ClickHouse的場景，我們也進行了查詢支持，ClickHouse支持TCP協議，http協議，mysql wire協議，目前Udata查詢引擎通過Mysql wire協議和ClickHouse進行外表關聯，

3.3.2 主要改造點介紹

• 在FE側，增加了JSF，HTTP，ClickHouse三種外部表對應的元資料結構，可以持久化外部表查詢需要的底層引擎的屬性資訊；
• FE側RBO改造，對于SQL語法樹對應的FromClause轉換為對應的邏輯計劃，并進一步轉換為物理計劃節點；
• BE側增加對應的ScanNode，進行資料查詢；
• 對于JSF和HTTP，通過函式，用于從FE側將查詢引數傳輸到BE側真實的查詢節點，查詢引數下推，同時列的過濾條件在獲取資料后，在Scan節點運行時過濾；
• 對于JSF和HTTP，建表中增加Mapping，將回傳的JSON資料映射到資料列；
• ClickHouse外部表查詢節點，可以支持兩種模式，普通的scan查詢和計算下推的Agg查詢；

3.3.3 使用方式及案例展示

1）Jsf外部表使用

Jsf建表陳述句 ( 表結構+訪問JSF必須的元資訊 )：

Mapping ( Jsf 回傳的json字串與資料表結構的映射 )：

查詢Sql陳述句 （ 查詢引數下推 和 列運算式運行時過濾 ）：

上面的sql是用來查詢Jsf外表的，同樣的其他聚合函式都可以用于該Jsf表查詢，上面主要有以下需要進行下說明:

1. 列運算式過濾：（ recv_count >= 1000 ) 這種過濾條件用于Scan操作獲取到資料之后，在BE節點內運行時進行再次過濾；
2. 查詢引數下推： jsfparam函式內置于Udata查詢引擎，可以通過此函式，將需要帶入到Jsf呼叫中的引數從呼叫端一直傳遞到Jsf服務中，從而減少資料的獲取；
3. 聯邦查詢：Jsf表同其他外表一樣可以支持聯邦查詢，也同樣可以支持其他外表支持的聚合等查詢操作；

2）Http外部表使用

Http建表陳述句：

Http的建表陳述句同上面Jsf表，只是Properties有所變化，變成了http訪問的元資訊，

查詢函式：

• httpconfig : 第一個引數是資料表中的某一個列名，后面是一個map, 目前僅支持 httpmethod 表示請求的方式 get/post ;
• httpheader : 第一個引數是資料表中的某一個列名，后面是一個map, json結構，決議后，按照key=>value 的配對，放入http請求的 header中去 ;
• httpbody : 第一個引數是資料表中的某一個列名，后面是引數，將直接放入http的請求的body中，這里需要注意的是 http請求的方式是 application/json , 還是 x-www-form-urlencoded ，兩種方式body中的寫法是不一樣的，x-www-form-urlencoded 寫法是 key1=value&key2=value2 ;

3.4 查詢代理-使Udata查詢引擎在理論上具備了查詢一切的可能性

UData查詢引擎目前支持的聯邦資料源有Es, Mysql, Ck, StarRocks, Hive, Iceberg, Hudi 等，同時對于UData目前不支持的資料源可以通過代理插件的形式進行擴展，我們提供了Udata Proxy的設計，只要遵循Udata代理提供的介面，實作對應的邏輯，來完成其他三方資料源的讀取，便可以集成到UData查詢引擎中，并和其他資料源一樣可以完成普通查詢和聯邦關聯查詢，

3.4.1 批處理 vs 分頁流式

Udata查詢引擎增加了Proxy scan 節點，Scan節點和Proxy代理之間可以通過Http和RPC兩種協議進行通訊；

資料從Proxy傳輸到Scan節點有兩種方式：

• 批處理：一次性獲取proxy回傳的全部資料；
• 分頁流式：適合資料量比較大的場景，利用scroll_id的引數，使資料可以分頁微批的方式流向scan節點，需要Proxy中邏輯代碼也支持滑動查詢；

3.4.2 邏輯讀插件熱插拔

任何異構的資料源可以通過邏輯讀插件的形式來支持，Proxy runtime 提供插件的執行環境，并進行并行執行緒的管理，邏輯讀插件可以通過Proxy管理端進行上傳和管理，熱插拔及時生效；

作者：劉敬斌 賀思遠

標籤：其他

上一篇：常見的排序演算法與時間復雜度

下一篇：2022 ICPC網路賽（二） G Good Permutation（樹形DP 排列組合 建樹）