當前Hadoop技術(shù)蓬勃發(fā)展，用于解決大數(shù)據(jù)的分析難題的技術(shù)平臺開始涌現(xiàn)。Spark憑借性能強勁、高度容錯、調(diào)度靈活等技術(shù)優(yōu)勢已漸漸成為主流技術(shù)，業(yè)界大部分廠商都提供了基于Spark的技術(shù)方案和產(chǎn)品。根據(jù)databricks的統(tǒng)計，目前有11個商業(yè)的Spark版本。
在使用Spark作出計算平臺的解決方案中，有兩種主流編程模型，一類是基于Spark API或者衍生出來的語言，另一種是基于SQL語言。SQL作為數(shù)據(jù)庫領域的事實標準語言，相比較用API(如MapReduce API，Spark API等)來構(gòu)建大數(shù)據(jù)分析的解決方案有著先天的優(yōu)勢：一是產(chǎn)業(yè)鏈完善，各種報表工具、ETL工具等可以很好的對接;二是用SQL開發(fā)有更低的技術(shù)門檻;三是能夠降低原有系統(tǒng)的遷移成本等。因此，SQL語言也漸漸成為大數(shù)據(jù)分析的主流技術(shù)標準。本文將深入解析Inceptor的架構(gòu)、編程模型和編譯優(yōu)化技術(shù)，并提供基準測試在多平臺上的性能對比數(shù)據(jù)。
1. Inceptor架構(gòu)
TranswARP Inceptor是基于Spark的分析引擎，如圖1所示，從下往上有三層架構(gòu)：最下面是存儲層，包含分布式內(nèi)存列式存儲(Transwarp Holodesk)，可建在內(nèi)存或者SSD上;中間層是Spark計算引擎層，星環(huán)做了大量的改進保證引擎有超強的性能和高度的健壯性;最上層包括一個完整的SQL 99和PL/SQL編譯器、統(tǒng)計算法庫和機器學習算法庫，提供完整的R語言訪問接口。
Transwarp Inceptor可以分析存儲在HDFS、Hbase或者Transwarp Holodesk分布式緩存中的數(shù)據(jù)，可以處理的數(shù)據(jù)量從GB到數(shù)十TB，即使數(shù)據(jù)源或者中間結(jié)果的大小遠大于內(nèi)存容量也可高效處理。另外Transwarp Inceptor通過改進Spark和YARN的組合，提高了Spark的可管理性。同時星環(huán)不僅僅是將Spark作為一個缺省計算引擎，也重寫了SQL編譯器，提供更加完整的SQL支持。
同時，Transwarp Inceptor還通過改進Spark使之更好地與Hbase融合，可以為Hbase提供完整的SQL支持，包括批量SQL統(tǒng)計、OLAP分析以及高并發(fā)低延時的SQL查詢能力，使得Hbase的應用可以從簡單的在線查詢應用擴展到復雜分析和在線應用結(jié)合的混合應用中，大大拓展了Hbase的應用范圍。
2. 編程模型
Transwarp Inceptor提供兩種編程模型：一是基于SQL的編程模型，用于常規(guī)的數(shù)據(jù)分析、數(shù)據(jù)倉庫類應用市場;二是基于數(shù)據(jù)挖掘編程模型，可以利用R語言或者Spark MLlib來做一些深度學習、數(shù)據(jù)挖掘等業(yè)務模型。
2.1 SQL模型
Transwarp Inceptor實現(xiàn)了自己的SQL解析執(zhí)行引擎，可以兼容SQL 99和HiveQL，自動識別語法，因此可以兼容現(xiàn)有的基于Hive開發(fā)的應用。由于Transwarp Inceptor完整支持標準的SQL 99標準，傳統(tǒng)數(shù)據(jù)庫上運行的業(yè)務可以非常方便的遷移到Transwarp Inceptor系統(tǒng)上。此外Transwarp Inceptor支持PL/SQL擴展，傳統(tǒng)數(shù)據(jù)倉庫的基于PL/SQL存儲過程的應用(如ETL工具)可以非常方便的在Inceptor上并發(fā)執(zhí)行。另外Transwarp Inceptor支持部分SQL 2003標準，如窗口統(tǒng)計功能、安全審計功能等，并對多個行業(yè)開發(fā)了專門的函數(shù)庫，因此可以滿足多個行業(yè)的特性需求。
2.2 數(shù)據(jù)挖掘計算模型
Transwarp Inceptor實現(xiàn)了機器學習算法庫與統(tǒng)計算法庫，支持常用機器學習算法并行化與統(tǒng)計算法并行化，并利用Spark在迭代計算和內(nèi)存計算上的優(yōu)勢，將并行的機器學習算法與統(tǒng)計算法運行在Spark上。例如：機器學習算法庫有包括邏輯回歸、樸素貝葉斯、支持向量機、聚類、線性回歸、關(guān)聯(lián)挖掘、推薦算法等，統(tǒng)計算法庫包括均值、方差、中位數(shù)、直方圖、箱線圖等。Transwarp Inceptor可以支持用R語言或者Spark API在平臺上搭建多種分析型應用，例如用戶行為分析、精準營銷、對用戶貼標簽、進行分類。
3. SQL編譯與優(yōu)化
Transwarp Inceptor研發(fā)了一套完整的SQL編譯器，包括HiveQL解析器、SQL標準解析器和PL/SQL解析器，將不同的SQL語言解析成中間級表示語言，然后經(jīng)過優(yōu)化器轉(zhuǎn)換成物理執(zhí)行計劃。SQL語言解析后經(jīng)過邏輯優(yōu)化器生成中間級表示語言，而中間表示語言再經(jīng)過物理優(yōu)化器生成最終的物理執(zhí)行計劃。從架構(gòu)上分，邏輯優(yōu)化器和物理優(yōu)化器都包含基于規(guī)則的優(yōu)化模塊和基于成本的優(yōu)化模塊。
為了和Hadoop生態(tài)更好的兼容，Inceptor為一個SQL查詢生成Map Reduce上的執(zhí)行計劃和Spark上的執(zhí)行計劃，并且可以通過一個SET命令在兩種執(zhí)行引擎之間切換。
3.1 SQL編譯與解析
Transwarp Inceptor的SQL編譯器會根據(jù)輸入的SQL查詢的類型來自動選擇不同的解析器，如PL/SQL存儲過程會自動進入PL/SQL解析器并生成一個Spark RDD的DAG從而在Spark平臺上并行計算，標準SQL查詢會進入SQL標準解析器生成Spark或Map Reduce執(zhí)行計劃。由于HiveQL和標準的SQL有所出入，為了兼容HiveQL，Transwarp Inceptor保留了HiveQL解析器，并可以對非標準SQL的Hive查詢生成Spark或者Map Reduce執(zhí)行計劃。
3.1.1 SQL 標準解析器
Transwarp Inceptor構(gòu)建了自主研發(fā)的SQL標準解析器，用于解析SQL 99 & SQL 2003查詢并生成Spark和Map Reduce的執(zhí)行計劃。詞法和語法分析層基于Antlr語法來構(gòu)建詞法范式，通過Antlr來生成抽象語義樹，并會通過一些上下文的語義來消除沖突并生成正確的抽象語義樹。語義分析層解析上層生成的抽象語義樹，根據(jù)上下文來生成邏輯執(zhí)行計劃并傳遞給優(yōu)化器。首先Transwarp Inceptor會將SQL解析成TABLE SCAN、SELECT、filter、JOIN、unio、ORDER BY、GROUP BY等主要的邏輯塊，接著會根據(jù)一些meta信息進一步細化各個邏輯塊的執(zhí)行計劃。如TABLE SCAN會分成塊讀取、塊過濾、行級別過濾、序列化等多個執(zhí)行計劃。
3.1.2 PL/SQL 解析器
PL/SQL是Oracle對SQL語言的模塊化擴展，已經(jīng)在很多行業(yè)中有大規(guī)模的應用，是數(shù)據(jù)倉庫領域的重要編程語言。
為了讓存儲過程在Spark上有較好的性能，PL/SQL解析器會根據(jù)存儲過程中的上下文關(guān)系來生成SQL DAG，然后對各SQL的執(zhí)行計劃生成的RDD進行二次編譯，通過物理優(yōu)化器將一些沒有依賴關(guān)系的RDD進行合并從而生成一個最終的RDD DAG。因此，一個存儲過程被解析成一個大的DAG，從而stage之間可以大量并發(fā)執(zhí)行，避免了多次執(zhí)行SQL的啟動開銷并保證了系統(tǒng)的并發(fā)性能。
解析并生成SQL級別的執(zhí)行計劃
解析SQL的依賴關(guān)系并生成DAG, 再根據(jù)各個SQL的執(zhí)行計劃來生成最終存儲過程的Spark RDD DAG
3.2 SQL優(yōu)化器
Transwarp Inceptor使用Spark作為默認計算引擎，并且開發(fā)了完善的SQL優(yōu)化器，因此在大量的客戶案例性能測試中，Transwarp Inceptor的性能領先Map Reduce 10-100倍，并超越部分開源MPP數(shù)據(jù)庫。SQL優(yōu)化器對平臺性能的提升居功至偉。
3.2.1 基于規(guī)則的優(yōu)化器(Rule based Optimizer)
目前為止，Transwarp Inceptor共實現(xiàn)了一百多個優(yōu)化規(guī)則，并且在持續(xù)的添加新的規(guī)則。按照功能劃分，這些規(guī)則主要分布在如下幾個模塊：
文件讀取時過濾
在文件讀取時過濾數(shù)據(jù)能夠最大化的減少參與計算的數(shù)據(jù)量從而最為有效的提高性能，因此Transwarp Inceptor提供了多個規(guī)則用于生成表的過濾條件。對于一些SQL中的顯示條件，Transwarp Inceptor會盡量將過濾前推到讀取表中;而對于一些隱式的過濾條件，如可以根據(jù)join key生成的過濾規(guī)則，Inceptor會根據(jù)語義保證正確性的前提下進行規(guī)則生成。
過濾條件前置
Transwarp Inceptor能夠從復雜的組合過濾條件中篩選出針對特定表的過濾規(guī)則，然后通過SQL語義來確定是否能將過濾條件前推到盡量早的時候執(zhí)行。如果有子查詢，過濾條件可以遞歸前推入最低層的子查詢中，從而保證所有的冗余數(shù)據(jù)被刪除。
超寬表的讀取過濾
對一些列超多的表進行處理的時候，Transwarp Inceptor首先會根據(jù)SQL語義來確定要讀取的列，并在讀取表的時候進行跨列讀取減少IO和內(nèi)存消耗。而如果表有過濾條件，Inceptor會做進一步優(yōu)化，首先只讀取過濾條件相關(guān)的列來確定該行記錄是否需要被選擇，如果不是就跳過當前行的所有列，因此能夠最大程度上的減少數(shù)據(jù)讀取。在一些商業(yè)實施中，這些優(yōu)化規(guī)則能夠帶來5x - 10x的性能提升。
Shuffle Stage的優(yōu)化與消除
Spark的shuffle實現(xiàn)的效率非常低，需要把結(jié)果寫磁盤，然后通過HTTP傳輸。Transwarp Inceptor添加了一些shuffle消除的優(yōu)化規(guī)則，對SQL的DAG中不必要或者是可以合并的shuffle stage進行消除或者合并。對于必須要做Shuffle的計算任務，Inceptor通過DAGScheduler來提高shuffle的效率：Map Task會直接將結(jié)果返回給DAGScheduler，然后DAGScheduler將結(jié)果直接交給Reduce Task而不是等待所有Map Task結(jié)束，這樣能夠非常明顯的提升shuffle階段的性能。
Partition消除
Transwarp Inceptor提供單一值Partition和Range Partition，并且支持對Partition建Bucket來做多次分區(qū)。當Partition過多的時候，系統(tǒng)的性能會因為內(nèi)存消耗和調(diào)度開銷而損失。因此，Inceptor提供了多個規(guī)則用于消除不必要的Partition，如果上下文中有隱式的對Partition的過濾條件，Inceptor也會生成對partition的過濾規(guī)則。
3.2.2 基于成本的優(yōu)化器(Cost based Optimizer)
基于規(guī)則的優(yōu)化器都是根據(jù)一些靜態(tài)的信息來產(chǎn)生的，因此很多和動態(tài)數(shù)據(jù)相關(guān)的特性是不能通過基于規(guī)則的優(yōu)化來解決，因此Transwarp Inceptor提供了基于成本的優(yōu)化器來做二次優(yōu)化。相關(guān)的原始數(shù)據(jù)主要來自meta-store中的表統(tǒng)計信息、RDD的信息、SQL上下文中的統(tǒng)計信息等。依賴于這些動態(tài)的數(shù)據(jù)，CBO會計算執(zhí)行計劃的物理成本并選擇最有效的執(zhí)行計劃。一些非常有效的優(yōu)化規(guī)則包括如下幾點：
JOIN順序調(diào)優(yōu)
在實際的案例中，join是消耗計算量最多的業(yè)務，因此對join的優(yōu)化至關(guān)重要。在多表JOIN模型中，Transwarp Inceptor會根據(jù)統(tǒng)計信息來預估join的中間結(jié)果大小，并選擇產(chǎn)生中間數(shù)據(jù)量最小的join順序作為執(zhí)行計劃。
JOIN類型的選擇
Transwarp Inceptor支持Left-most Join Tree 和 Bush Join Tree，并且會根據(jù)統(tǒng)計信息來選擇生成哪種Join模型有最佳性能。此外，Transwarp Inceptor會根據(jù)原始表或者中間數(shù)據(jù)的大小來選擇是否開啟針對數(shù)據(jù)傾斜模型下的特殊優(yōu)化等。此外，針對Hbase表是否有索引的情況，Transwarp Inceptor會在普通Join和Look-up Join間做個均衡的選擇。
并發(fā)度的控制
Spark通過線程級并發(fā)來提高性能，但是大量的并發(fā)可能會帶來不必要的調(diào)度開銷，因此不同的案例在不同并發(fā)度下會有最佳性能。Transwarp Inceptor通過對RDD的一些屬性進行推算來選擇最佳并發(fā)控制，對很多的案例有著2x-3x的性能提升。
4.Transwarp Holodesk內(nèi)存計算引擎
為了有效的降低SQL分析的延時，減少磁盤IO對系統(tǒng)性能的影響，星環(huán)科技研發(fā)了基于內(nèi)存或者SSD的存儲計算引擎Transwarp Holodesk，通過將表數(shù)據(jù)直接建在內(nèi)存或者SSD上以實現(xiàn)SQL查詢?nèi)珒?nèi)存計算。另外Transwarp Holodesk增加了數(shù)據(jù)索引功能，支持對多個數(shù)據(jù)列建索引，從而更大程度的降低了SQL查詢延時。
4.1 存儲格式
Transwarp Holodesk基于列式存儲做了大量的原創(chuàng)性改進帶來更高的性能和更低的數(shù)據(jù)膨脹率。首先數(shù)據(jù)被序列化后存儲到內(nèi)存或SSD上以節(jié)省者資源占用。如圖3所示，每個表的數(shù)據(jù)被存儲成若干個Segment，每個Segment被劃分成若干個Block，每個Block按照列方式存儲于SSD或內(nèi)存中。另外每個Block的頭部都加上Min-Max Filter和Bloom Filter用于過濾無用的數(shù)據(jù)塊，減少不必要的數(shù)據(jù)進入計算階段。
Transwarp Holodesk根據(jù)查詢條件的謂詞屬性對每個數(shù)據(jù)塊的對應列構(gòu)建數(shù)據(jù)索引，索引列采用自己研發(fā)的TrIE結(jié)構(gòu)進行組織存儲，非索引列采用字典編碼的方式進行組織存儲。Trie不僅能對具有公共前綴的字符串進行壓縮，而且可以對輸入的字符串排序，從而可以利用二分查找快速查詢所需數(shù)據(jù)的位置，從而快速響應查詢需求。
HDFS 2.6支持Storage Tier讓應用程序可以選擇存儲層為磁盤或者SSD，但是沒有專用的存儲格式設計是無法有效利用SSD的讀寫吞吐量和低延，因此現(xiàn)有的Text以及行列混合(ORC/Parquet)都不能有效的利用SSD的高性能。為此驗證存儲結(jié)構(gòu)對性能的影響，我們將HDFS構(gòu)建在SSD上并選用某基準測試來做了進一步的性能對比，結(jié)果如圖4所示：采用文本格式，PCI-E SSD帶來的性能提升僅1.5倍;采用專為內(nèi)存和SSD設計的Holodesk列式存儲，其性能相比較SSD上的HDFS提升高達6倍。
4.2 性能優(yōu)勢
某運營商客戶在12臺x86服務器上搭建了Transwarp Inceptor，將Transwarp Holodesk 配置在PCIE-SSD上，并與普通磁盤表以及DB2來做性能對比測試。最終測試數(shù)據(jù)如圖5所示：
在純粹的count測試一項，Holodesk性能相對于磁盤表最高領先32倍;對于join測試一項，Transwarp Holodesk最高領先磁盤表多達12倍;在單表聚合測試中，Holodesk提升倍數(shù)達10~30倍。另外Transwarp Holodesk在和DB2的對比中也表現(xiàn)優(yōu)秀，兩個復雜SQL查詢在DB2數(shù)據(jù)庫中需要運行1小時以上，但是在使用Transwarp Holodesk均是分鐘級和秒級就返回結(jié)果。
內(nèi)存的價格大約是同樣容量SSD的十倍左右，為了給企業(yè)提供更高性價比的計算方案，Transwarp Holodesk針對SSD進行了大量的優(yōu)化，使得應用在SSD上運行具有與在內(nèi)存上比較接近的性能，從而為客戶提供了性價比更高的計算平臺。
在對TPC-DS的IO密集型查詢的測試中，無論上構(gòu)建在PCI-E SSD還是內(nèi)存上，Holodesk對比磁盤表有一個數(shù)量級上的性能提升;而SSD上的Holodesk性能只比內(nèi)存差10%左右。
5. 穩(wěn)定的Spark執(zhí)行引擎
企業(yè)目前應用開源Spark的主要困難在穩(wěn)定性、可管理性和功能不夠豐富上。開源Spark在穩(wěn)定性上還有比較多的問題，在處理大數(shù)據(jù)量時可能無法運行結(jié)束或出現(xiàn)Out of memory，性能時快時慢，有時比Map/Reduce更慢，無法應用到復雜數(shù)據(jù)分析業(yè)務中。
Transwarp Inceptor針對各種出錯場景設計了多種解決方法，如通過基于成本的優(yōu)化器選擇最合適的執(zhí)行計劃、加強對數(shù)據(jù)結(jié)構(gòu)內(nèi)存使用效率的有效管理、對常見的內(nèi)存出錯問題通過磁盤進行數(shù)據(jù)備份等方式，極大提高了Spark功能和性能的穩(wěn)定性，上述問題都已經(jīng)解決并經(jīng)過商業(yè)案例的考驗。Transwarp Inceptor能穩(wěn)定的運行7*24小時，并能在TB級規(guī)模數(shù)據(jù)上高效進行各種穩(wěn)定的統(tǒng)計分析。
6. SQL引擎效能驗證
TPC-DS是TPC組織為Decision Support System設計的一個測試集，包含對大數(shù)據(jù)集的統(tǒng)計/報表生成/聯(lián)機查詢/數(shù)據(jù)挖掘等復雜應用，測試用的數(shù)據(jù)有各種不同的分布與傾斜，與真實場景非常接近。隨著國內(nèi)外各代表性的Hadoop發(fā)行版廠商以TPC-DS為標準測評產(chǎn)品，TPC-DS也就逐漸成為了業(yè)界公認的Hadoop系統(tǒng)測試準則。
6.1 驗證對比的平臺和配置
我們搭建了兩個集群分別用于Transwarp Inceptor與Cloudera data hub/Impala的測試。每個集群采用4臺普通兩路x86服務器搭建，每臺服務器硬件配置如下：
考慮到磁盤的容量和HDFS的存儲復制模式，我們選擇的是500GB的數(shù)據(jù)總量。SQL測試案例的選擇上，在Cloudera Impala中使用的是由Cloudera改動過的TPC-DS測試子集，在Transwarp Inceptor我們選用的是TPC-DS為Oracle生成的測試集合，保留了原有的各種復雜SQL，因此能夠客觀反映出Inceptor在SQL支持上的情況。
6.2 Transwarp Inceptor VS Cloudera Impala
Transwarp Inceptor由于有完善的SQL支持，能夠運行全部所有的99個SQL查詢。而由于Cloudera官方發(fā)布的TPC-DS測試集只包含19個SQL案例，因此我們只能運行這19個SQL，實驗證明這部分查詢在Impala上全部正常運行完成。
圖7是所有的測試集合的性能對比圖。圖中縱坐標小于1表述測試案例中Cloudera Impala性能超過Transwarp Inceptor，而大于1則表示Transwarp Inceptor有更好的性能表現(xiàn)。對于Cloudera Impala不能支持的SQL，我們就標記這個性能比為100。從圖中可見，在Cloudera Impala支持的19個SQL中，有8個SQL的表現(xiàn)超過Transwarp Inceptor，2個表現(xiàn)相當，另外9個Transwarp Inceptor比Cloudera Impala表現(xiàn)的更好。
6.3 Transwarp Inceptor VS Map Reduce
我們使用了同樣的硬件和軟件配置完成和開源的Hive執(zhí)行效率相比，Transwarp Inceptor能夠帶來10x-100x的性能提升。圖8是TPC-DS的部分SQL查詢在Inceptor和CDH 5.1 Hive的性能提升倍數(shù)，其中最大的提升倍數(shù)竟可達到123倍。
7. 結(jié)語
隨著在大數(shù)據(jù)領域國內(nèi)外開始處于同一起跑線，我們相信像星環(huán)科技這樣國內(nèi)具有代表性的Hadoop發(fā)行版廠商將在中國的廣闊市場空間中獲得長足發(fā)展，并且由于中國市場激烈的競爭與磨練，逐步打磨出超越國外先進廠商的技術(shù)與實力。
劉汪根。2013年加入星環(huán)，作為早期員工參與了星環(huán)大數(shù)據(jù)平臺的構(gòu)建，現(xiàn)擔任數(shù)據(jù)平臺部研發(fā)經(jīng)理，主要負責與管理星環(huán)大數(shù)據(jù)平臺數(shù)據(jù)平臺的研發(fā)工作，如SQL編譯器，Spark執(zhí)行引擎等工作，產(chǎn)品涵括Transwarp Inceptor/Transwarp Stream等軟件。
聯(lián)系客服
