EMR 打造高效雲原生數據分析引擎

本場視頻連接：EMR打造高效雲原生數據分析引擎

本場ppt材料：https://www.slidestalk.com/AliSpark/2019___0926_110365

---

基於開源體系打造雲上數據分析平臺

客戶選擇開源方案的緣由主要有如下幾點：

• 靈活多樣的業務場景：目前即使是一個小企業，其數據存儲也多是多種多樣的，好比業務數據、日誌數據和圖數據等，這種狀況下，須要有一個高度定製化的系統來串聯不一樣的業務場景；
• 本身有專業的運維能力：開源系統有充足的人才儲備，豐富的網上資料與開源的強大後盾，能夠確保公司業務的順利開展；
• 多種業務需求vs成本壓力：每種雲上產品有本身的使用場景，對於中小企業來講購買多種雲產品將會形成很大的成本壓力，而經過開源體系維護一套系統，在集成用戶業務中所需組件的同時，能夠下降用戶的成本。

下圖是阿里巴巴EMR系統的產品架構圖。用戶上雲的方式主要有兩種，一種是購買ECS資源本身搭建一套開源系統；另外一種是直接選擇阿里巴巴的EMR系統。第一種方式因爲開源系統組件多，涉及到了Spark、Hive、Flink和TensorFlow等，從零搭建一套完整的大數據系統對於用戶來說很是複雜，尤爲是成百上千的集羣規模也給運維形成了很大的挑戰。而使用EMR系統具備如下優勢：
1) 阿里雲EMR系統能夠幫助用戶一鍵化自動部署、配置相關組件，開箱即用，同時還會根據用戶的機器類型進行參數的自動推薦調優。
2) 阿里雲EMR系統與阿里雲其餘產品實現了打通，好比數據存放在OSS，EMR系統無需額外再作認證配置，即可以很方便地讀取OSS上的數據；
3) 阿里雲EMR系統集成了不少自研插件，這些插件在其餘產品中是沒有的；
4) 阿里雲EMR系統的全部組件兼容開源但優於開源，如Flink集成了阿里雲自研的Blink和TensorFlow（PAI），這也是阿里云爲社區作的一點貢獻，目的是爲了讓用戶能用到阿里雲內部的技術；
5) 阿里雲EMR系統提供了全平臺的做業診斷與告警組件APM來實現自動化運維，大大下降集羣運維的複雜性；
6) 阿里雲EMR系統還與DataWorks對接，用戶能夠以DataWorks爲入口，傻瓜式地使用EMR系統。

EMR系統的目標主要有如下三個：

• 平臺化：將EMR作成一個統一的雲上數據分析平臺，幫助用戶打造全棧式的大數據解決方案，支持全系列VM容器化，提供企業級HAS和大數據APM；
• 技術社區&深度：持續深耕技術社區，打造大數據友好的雲 Native 存儲，同時將技術回饋給社區，爲社區作貢獻；
• 生態：EMR系統將結合阿里雲其餘產品構建一個生態，接入Blink、PAI，集成OSS、OTS方案。

EMR-Jindo：雲原生高效數據分析引擎

下圖展現了TPC-DS的基準測試報告，能夠發如今2019年3月份10TB的測試中，性能指標得分是182萬左右，成本是0.31 USD；而2019年十月份一樣的測試性能指標得分已經變成526萬，成本降低到0.53 CNY，也就是說通過半年左右性能提高了2.9倍，成本縮減到原來的四分之一。同時阿里巴巴還成爲了首個提交TPC-DS測試100TB測試報告的廠商。這些成績的背後是EMR-Jindo引擎的支持。

EMR-Jindo引擎架構主要分爲兩部分：

• Jindo-Spark：EMR內部全面優化的Spark高效計算引擎，能夠處理多種計算任務；
• Jindo-FS：自研的雲原生存儲引擎，兼容開源HDFS的接口，兼顧性能與價格。

1） Jindo-Spark

Jindo-Spark高效計算引擎對Spark採起了一系列優化措施，好比Runtime Filter支持自適應的運行時數據裁剪；Enhanced Join Reorder來解決外鏈接重排等問題；TopK支持推理並下推 TopK 邏輯，幫助儘早地過濾數據；File Index支持文件級別過濾和min/max/bloom/倒排等；自研開發了Relational Cache，實現使用一套引擎就能夠將查詢從分鐘級提高爲亞秒級；針對特定的場景推出Spark Transaction功能，爲Spark引入Full ACID支持；實現了Smart Shuffle功能，從底層來減小sort-merge 次數，提高Shuffle的效率。

• Runtime Filter：
相似於Spark中的Dynamic Partition Pruning（DPP），可是其比DPP功能更強大。除了DPP能處理的分析表以外，Runtime Filter還能夠處理非分析表。其基本原理是運行時動態裁剪數據，避免沒必要要的計算。好比，面對一個join查詢，沒法經過value下推到存儲層而將數據過濾，邏輯推算的時候沒法預知最後的數據量級。這種狀況下若是是分析表，Runtime Filter首先會估計其中一個表中參與join操做的數據量，若是數據量較小，則提早進行數據篩選，再推送到另外一側作數據過濾；而對於非分析表，會引入Filter，如BloomFilter得到Min或Max的統計信息，

根據這些統計信息，將備選數據比較少的一側提取出來，推到另外一側進行過濾。Runtime Filter的成本很小，只須要在優化器中進行簡單評估，卻能夠帶來顯著的性能提高。以下圖所示，Runtime Filter實現了35%左右的總體性能提高。該特性已經在Spark提交了PR（SPARK-27227）。
• Enhanced Join Recorder：
你們都知道，算子執行順序可能會極大地影響sql的執行效率，這種狀況下優化的核心原則是改變算子的執行順序，儘早地過濾數據。

好比下圖左上角的例子中，若是最底層兩個表很是大的話，則這兩張表join的開銷會很是大，join後的大數據再去join小表，大數據一層一層地傳遞下去，就會影響整個流程的執行效率。此時，優化的思想是先將大表中一些無關的數據過濾掉，減小往下游傳遞的數據量。針對該問題，Spark使用的是動態規劃算法，但其只適用於表的數量比較少的狀況，若是表的數量大於12，該算法就一籌莫展。面對表的數量比較多的狀況，EMR提供了多表join的遺傳算法，其能夠將原來的動態規劃算法的2n的複雜度降到線性的量級，能完成成百上千張表的join。

下圖右上角能夠看到，Query64有18個表參與join，動態規劃算法優化時間就須要耗費1400秒，而多表join的遺傳算法僅須要20秒左右就可完成。Join Recorder另一個重要的功能是外鏈接重排算法，你們都知道sql中外鏈接不能隨意交換順序的，但這並不表明不能交換順序，好比A left join B， 而後再left join C，事實上在某種條件下其順序是可交換的。在Spark中，外鏈接的優化是直接被放棄掉，而EMR則根據現有研究找到了順序可交換的充分必要條件，實現了外鏈接重排算法（以下圖左下角所示），對外鏈接的執行效率有了質的提高（下圖右下角）

• Relational Cache：
Spark本來的Cache存在幾個侷限點，其一Spark的Cache是session級別，若是發現某一個Query的片斷使用比較頻繁，就會對爲這個session建立一個cache，可是session結束後，cache就會消失；其二Spark的Cache是存儲在本機上，而不是分佈式存儲，所以沒法作到通用。在此基礎上，EMR平臺實現了Relational Cache，對任意Spark表，視圖或者Dataset等關係型數據抽象的數據實體都建立cache, 相似於物化視圖（Materialized View），可是比物化視圖功能要豐富。Relational Cache的使用場景包括a）亞秒級響應MOLAP引擎;b）交互式BI，Dashboard；c）數據同步；d）數據預組織。

Relational Cache的建立過程以下，其語法與Spark sql常見的DDL相似。首先CACHE一個表或視圖，而後指定Relational Cache的更新策略（DEMAND或COMMIT）、是否用於後續優化、Cache數據的存儲方式以及Cache的視圖邏輯。Relational Cache支持cache任意Table、View，支持cache到內存、HDFS、OSS等任意數據源，JSON、ORC、Parquet等任意數據格式。

Relational Cache還支持對用戶輸入的sql的優化。原來的Spark sql Cache對於用戶輸入的sql優化很是僵硬死板，用戶輸入的sql必須精確匹配上Cache才能使用。而Relational Cache則徹底不一樣，若是有a、b、c、d四個表join的cache，當又有a、b、e三個表join的狀況下，a、b join的結果即可以從四個表join時生成的Cache數據中讀取。下圖中右側展現了Cache和沒有Cache的基準測試結果，能夠看出Relational Cache能夠保證測試的響應時間在亞秒級。
請參考 Spark Relational Cache實現亞秒級響應的交互式分析

• Spark Transaction：有些用戶可能會使用Hive表，Hive表有事務支持，然而Spark在事務這一項上是不兼容Hive的。所以，爲了知足用戶數據訂正/刪除以及數據流導入的場景支持，EMR平臺提供了Spark Transaction支持事務的ACID支持。

傳統的數據導入是分批的，好比一天一導入，而流數據導入場景下數據是實時寫入的原始數據，並未通過任何處理，所以會有delete和update的需求。Spark Transaction總體來說是一種鎖+MVCC的實現形式，MVCC與底層的存儲密不可分。大數據在Hive和Spark兼容的狀況下，都是文件的形式存在目錄中，文件的版本經過行來控制，寫入的每一行都會加上Meta Columns，如op、original_write-id、bucket id和row_id等，來標識這是全表惟一的一行。當須要更新某一行的時候，並不會原地更新該行，而是將該行取出來，重寫後產生新的版本進行存儲。讀取的時候，多版本會進行合併後返回給用戶。

###2） Jindo-FS
EMR早期推出了一種本地盤機型，使用這種機型來部署集羣相似於用本地集羣在雲下部署大數據發行版，價格較高；此外因爲當時HDFS有元數據瓶頸，本地存儲的動態化伸縮面臨很大的挑戰。針對這方面的問題，解決的方案是計算與存儲分離，將數據存儲在OSS上，可是這種分離帶來的直接結果就是性能變差，由於OSS元數據操做耗時，讀取數據跨網絡，傳輸帶寬也會嚴重影響性能。

進而的解決方案是將數據從遠端拉取到計算側進行緩存，這也是Jindo-FS作的事情。Jindo-FS是相似於HDFS的系統，其架構也相似於HDFS的Master-Slave架構，分紅Name Service 和Storage Service。它支持將某些訪問頻率比較高的表能夠放到RocksDB中進行多級緩存。Jindo-FS總體不一樣於HDFS的Master結點， Jindo-FS的「Master」（Name Service）是一個分佈式集羣，使用raft 協議，提供入口服務；提供多Name Space支持；元數據以kv形式存放於高性能kv store 中；由於其自己不存儲數據，真實數據在OSS和OTS中，所以支持數據的彈性擴展和銷燬重建。

Jindo-FS底層的元數據管理會將數據拆成一系列的kv，經過遞增的id來逐層查詢。如/home/Hadoop/file1.txt須要讀三次OTS。下圖右側的測試結果說明Jindo-FS在元數據操做方面相對於OSS有較好的性能提高。

Jindo-FS使用Storage Service來進行底層存儲，在寫流程中Storage Service將要寫的文件同時存儲到本地和OSS中，而後再返回給用戶寫的結果，同時還會在集羣結點內進行多副本傳輸；而讀操做和HDFS相似，若是命中本地，則在本地存儲中讀取，不然要進行遠程讀取。Storage Service具有高性能、高可靠、高可用、彈性存儲等特性，爲了支持高性能，Jindo-FS創建了數據流高速通道，同時還有一系列的策略，如減小內存拷貝次數等。

Jindo-FS中Name Service如何實現高可靠、如何進行熱點數據發現與緩存替換、塊存儲模式與緩存模式；以及Storage Service如何應對讀寫失敗、數據塊如何設計並存儲、如何實現高速數據通道等問題，請參見大數據生態專場《雲上大數據的高效能數據庫的存儲方案》的分享。

相關文章：
JindoFS概述：雲原生的大數據計算存儲分離方案

JindoFS解析 - 雲上大數據高性能數據湖存儲方案

 

雙11福利來了！先來康康#怎麼買雲服務器最便宜# [並不簡單]參團購買指定配置雲服務器僅86元/年，開團拉新享三重禮：1111紅包+瓜分百萬現金+31%返現，爆款必買清單，還有iPhone 11 Pro、衛衣、T恤等你來抽，立刻來試試手氣 https://www.aliyun.com/1111/2019/home?utm_content=g_1000083110

 

閱讀原文

本文爲雲棲社區原創內容，未經容許不得轉載。