2014年，Spark開(kāi)源生態(tài)系統得到了大幅增長(cháng)，已成為大數據領(lǐng)域最人氣的開(kāi)源項目之一，活躍在Hortonworks、IBM、Cloudera、MapR和Pivotal等眾多知名大數據公司，更擁有Spark SQL、Spark Streaming、MLlib、GraphX等多個(gè)相關(guān)項目。同時(shí)值得一提的是，Spark貢獻者中有一半左右的中國人。
短短四年時(shí)間，Spark不僅發(fā)展為Apache基金會(huì )的頂級開(kāi)源項目，更通過(guò)其高性能內存計算及其豐富的生態(tài)快速贏(yíng)得幾乎所有大數據處理用戶(hù)。2015年1月10日，一場(chǎng)基于Spark的高性能應用實(shí)踐盛宴由Databricks軟件工程師連城、百度高級工程師甄鵬、百度架構師孫垚光、百度美國研發(fā)中心高級架構師劉少山四位專(zhuān)家聯(lián)手打造。
Databricks軟件工程師連城——Spark SQL 1.2的提升和新特性
談及Spark SQL 1.2的提升和新特性，連城主要總結了4個(gè)方面——External data source API（外部數據源API）、列式內存存儲加強（Enhanced in-memory columnar storage）、Parquet支持加強（Enhanced Parquet support）和Hive支持加強（Enhanced Hive support）。
External data source API
連城表示，因為在處理很多外部數據源中出現的擴展問(wèn)題，Spark在1.2版本發(fā)布了External data source API。通過(guò)External data source API，Spark將不同的外部數據源抽象成一個(gè)關(guān)系表格，從而實(shí)現更貼近無(wú)縫的操作。
External data source API在支持了多種如JSON、Avro、CSV等簡(jiǎn)單格式的同時(shí)，還實(shí)現了Parquet、ORC等的智能支持；同時(shí)，通過(guò)這個(gè)API，開(kāi)發(fā)者還可以使用JDBC將HBase這樣的外部系統對接到Spark中。
連城表示，在1.2版本之前，開(kāi)發(fā)者其實(shí)已經(jīng)實(shí)現了各種各樣外部數據源的支持，因此，對比更原生的支持一些外部數據源，External data source API的意義更在于針對相應數據源進(jìn)行的特殊優(yōu)化，主要包括Column pruning（列剪枝）和Pushing predicates to datasources（將predicates貼近數據源）兩個(gè)方面：
Column pruning。主要包括縱橫的兩種剪枝。在列剪枝中，Column pruning可以完全忽視無(wú)需處理的字段，從而顯著(zhù)地減少I(mǎi)O。同時(shí)，在某些條件查詢(xún)中，基于Parquet、ORC等智能格式寫(xiě)入時(shí)記錄的統計信息（比如最大值、最小值等），掃描可以跳過(guò)大段的數據，從而省略了大量的磁盤(pán)掃描負載。
Pushing predicates to datasources。在更復雜的SQL查詢(xún)中，讓過(guò)濾條件維度盡可能的接近數據源，從而減少磁盤(pán)和網(wǎng)絡(luò )IO，最終提高整體端到端的性能。
使用External data source API之前
使用External data source API之后
搭載了如Parquet和ORC這樣的智能格式
連城表示，在Spark 1.2版本中，External data source API并沒(méi)有實(shí)現預期中的功能，在Roadmap中，First class分片支持（First class partitioning support with partition pruning）、Data sink（insertion）API、將Hive作為外部數據源等。
Enhanced in-memory columnar storage
連城表示，不管Shark，還是Spark，內存緩存表的支持都是非常重要的一個(gè)特性。他表示，雖然在1.1和之前版本中的列式內存表的性能已然不錯，但是還會(huì )出現一些問(wèn)題：第一，大數據量下緩存超大體積表時(shí)（雖然不推薦，但不缺現實(shí)用例），會(huì )出現OOM等問(wèn)題；第二，在列式存儲中，像Parquet、ORC這種收集統計信息然后通過(guò)這些信息做partition skipping等操作在之前版本中并沒(méi)有完全實(shí)現。這些問(wèn)題在1.2版本中都得到了解決，本節，連城主要介紹了語(yǔ)義統一、緩存實(shí)體化、基于緩存共享的查詢(xún)計劃、Cache大表時(shí)的OOM問(wèn)題、表格統計（Table statistics）等方面。
緩存實(shí)體化。SQLContext.cacheTable（“tbl”）默認使用eager模式，緩存實(shí)體化將自動(dòng)進(jìn)行，不會(huì )再等到表被使用或觸發(fā)時(shí)，避免手動(dòng)做“SELECT COUNT（*） FROM src;”。同時(shí)，新增了“CACHE [LAZY] TABLE tbl [AS SELECT …]”這樣的DML。
語(yǔ)義統一。早期時(shí)候，SchemaRDD.cache（）和SQLContext.cacheTable（“tbl”）這兩個(gè)語(yǔ)義是不同的。其中，SQLContext.cacheTable會(huì )去建立一些列式存儲格式相關(guān)優(yōu)化，而SchemaRDD.cache（）卻以一行一個(gè)對象的模式進(jìn)行。在1.2版本中，這兩個(gè)操作已被統一，同時(shí)各種cache操作都將得到一個(gè)統一的內存表。
基于緩存共享的查詢(xún)計劃。兩個(gè)得到相同結果的cache語(yǔ)句將共享同一份緩存數據。
避免Cache大表時(shí)的OOM問(wèn)題。優(yōu)化內存表的建立和訪(fǎng)問(wèn)，減少開(kāi)銷(xiāo)，進(jìn)一步提升性能；在緩存大表時(shí)，引入batched column buffer builder，將每一列切成多個(gè)batch，從而避免了OOM。
表格統計。Table statistics，類(lèi)似Parquet、ORC使用的技術(shù)，在1.2版本中主要實(shí)現了Predicate pushdown（實(shí)現更快的表格掃描）和Auto broadcast join（實(shí)現更快的表格join）。
最后，連城還詳細介紹了一些關(guān)于加強Parquet和Hive支持的實(shí)現，以及Spark未來(lái)的一些工作。
百度基礎架構部高級工程師甄鵬——Spark在百度開(kāi)放云BMR中的實(shí)戰分享
百度分布式計算團隊從2011年開(kāi)始持續關(guān)注Spark，并于2014年將Spark正式引入百度分布式計算生態(tài)系統中，在國內率先面向開(kāi)發(fā)者及企業(yè)用戶(hù)推出了支持Spark并兼容開(kāi)源接口的大數據處理產(chǎn)品BMR（Baidu MapReduce）。在甄鵬的分享中，我們主要了解了百度Spark 應用現狀、百度開(kāi)放云BMR和Spark On BMR三個(gè)方面的內容。
Spark在百度
甄鵬表示，當前百度的Spark集群由上千臺物理主機（數萬(wàn)Cores，上百TBMemory）組成，日提交App在數百，已應用于鳳巢、大搜索、直達號、百度大數據等業(yè)務(wù)。之以選擇Spark，甄鵬總結了三個(gè)原因：快速高效、API 友好易用和組件豐富。
快速高效。首先，Spark使用了線(xiàn)程池模式，任務(wù)調度效率很高；其次，Spark可以最大限度地利用內存，多輪迭代任務(wù)執行效率高。
API友好易用。這主要基于兩個(gè)方面：第一，Spark支持多門(mén)編程語(yǔ)言，可以滿(mǎn)足不同語(yǔ)言背景的人使用；第二，Spark的表達能力非常豐富，并且封裝了大量常用操作。
組件豐富。Spark生態(tài)圈當下已比較完善，在官方組件涵蓋SQL、圖計算、機器學(xué)習和實(shí)時(shí)計算的同時(shí)，還有著(zhù)很多第三方開(kāi)發(fā)的優(yōu)秀組件，足以應對日常的數據處理需求。
百度開(kāi)放云BMR
在BMR介紹中，甄鵬表示，雖然BMR被稱(chēng)為Baidu MapReduce，但是這個(gè)名稱(chēng)已經(jīng)不能完全表示出這個(gè)平臺：BMR是百度開(kāi)放云的數據分析服務(wù)產(chǎn)品，基于百度多年大數據處理分析經(jīng)驗，面向企業(yè)和開(kāi)發(fā)者提供按需部署的Hadoop&Spark集群計算服務(wù)，讓客戶(hù)具備海量數據分析和挖掘能力，從而提升業(yè)務(wù)競爭力。
如圖所示，BMR基于BCC（百度云服務(wù)器），建立在HDFS和BOS（百度對象存儲）分布式存儲之上，其處理引擎包含了MapReduce和Spark，同時(shí)還使用了HBase數據庫。在此之上，系統集成了Pig、Hive、SQL、Streaming、GraphX、MLLib等專(zhuān)有服務(wù)。在系統的最上層，BMR提供了一個(gè)基于Web的控制臺，以及一個(gè)API形式的SDK。
在圖片的最右邊，Scheduler在BMR中起到了管理作用，使用它開(kāi)發(fā)者可以編寫(xiě)比較復雜的作業(yè)流。
Spark On BMR
類(lèi)似于通常的云服務(wù)，BMR中的Spark同樣隨用隨起，集群空閑即銷(xiāo)毀，幫助用戶(hù)節省預算。此外，集群創(chuàng )建可以在3到5分鐘內完成，包含了完整的Spark+HDFS+YARN堆棧。同時(shí)，BMR也提供Long Running模式，并有多種套餐可選。
完善的報表服務(wù)，全方位監控
在安全上，用戶(hù)擁有虛擬的獨立網(wǎng)絡(luò )，在同一用戶(hù)全部集群可互聯(lián)的同時(shí)，BMR用戶(hù)間網(wǎng)絡(luò )被完全隔離。同時(shí)，BMR還支持動(dòng)態(tài)擴容，節點(diǎn)規模可彈性伸縮。除此之外，在實(shí)現Spark全組件支持的同時(shí)，BMR可無(wú)縫對接百度的對象存儲BOS服務(wù)，借力百度多年的存儲研發(fā)經(jīng)驗，保證數據存儲的高可靠性。
百度基礎架構部架構師孫垚光——百度高性能通用Shuffle服務(wù)
在2014 Sort Benchmark國際大賽上，百度成功奪冠，其幕后英雄無(wú)疑卓越的Shuffle機制，在孫垚光的分享中，我們對Shuffle的發(fā)展、細節和未來(lái)有了一次深度的接觸。
Shuffle簡(jiǎn)介
孫垚光表示，簡(jiǎn)單來(lái)說(shuō)，Shuffle就是按照一定的分組和規則Map一個(gè)數據，然后傳入Reduce端。不管對于MapReduce還是Spark，Shuffle都是一個(gè)非常重要的階段。然而，雖然Shuffle解決的問(wèn)題相同，但是在Spark和MapReduce中，Shuffle流程（具體時(shí)間和細節）仍然存在一定的差別：
Baidu Shuffle發(fā)展歷程
通過(guò)孫垚光了解到，Shuffle在百度的發(fā)展主要包括兩個(gè)階段：跟隨社區和獨立發(fā)展。從2008年百度的MapReduce/Hadoop起步開(kāi)始，百度就開(kāi)始跟隨社區，使用社區版本，期間的主要工作包含Bug修復和性能優(yōu)化兩個(gè)方面（增加內存池、減少JVMGC，傳輸Server由Jetty換Netty，及批量傳輸、聚合數據等方面）。
分離了shuffle和Map/Reduce
在2012年開(kāi)始，Baidu Shuffle開(kāi)啟獨立發(fā)展階段，主要源于下一代離線(xiàn)計算系統的開(kāi)發(fā)，Shuffle被抽離為獨立的ShuffleService服務(wù)，從而提高了集群資源的利用率。
截止此時(shí)，不管是社區版本（MapReduce/Spark），還是百度研發(fā)的ShuffleService，它們都是基于磁盤(pán)的PULL模式。基于磁盤(pán)，所有Map的數據都會(huì )放到磁盤(pán)，雖然Spark號稱(chēng)內存計算，但是涉及到Shuffle時(shí)還是會(huì )寫(xiě)磁盤(pán)。基于PULL，所有數據在放到Map端的磁盤(pán)之后，Reduce在使用時(shí)還需要主動(dòng)的拉出來(lái)，因此會(huì )受到兩個(gè)問(wèn)題影響：首先，業(yè)務(wù)數據存儲在Map端的服務(wù)器上，機器宕機時(shí)會(huì )不可避免丟失數據，這一點(diǎn)在大規模分布式集群中非常致命；其次，更重要的是，Shuffle階段會(huì )產(chǎn)生大量的磁盤(pán)尋道（隨機讀）和數據重算（中間數據存在本地磁盤(pán)），舉個(gè)例子，某任務(wù)有1百萬(wàn)個(gè)Map，1萬(wàn)個(gè)Reduce，如果一次磁盤(pán)尋道的時(shí)間是10毫秒，那么集群總共的磁盤(pán)尋道時(shí)間= 1000000 ×10000 ×0.01 = 1億秒。
New Shuffle
基于這些問(wèn)題，百度設計了基于內存的PUSH模式。新模式下，Map輸出的數據將不落磁盤(pán)，并在內存中及時(shí)地Push給遠端的Shuffle模塊，從而將獲得以下提升：
New Shuffle的優(yōu)勢
New Shuffle架構
如圖所示，藍色部分為New Shuffle部分，主要包含兩個(gè)部分：數據寫(xiě)入和讀取的API，Map端會(huì )使用這個(gè)接口來(lái)讀取數據，Reduce會(huì )使用這個(gè)接口來(lái)讀取數據；其次，最終重要的是，服務(wù)器端使用了典型的主從架構，用多個(gè)shuffle工作者節點(diǎn)來(lái)shuffle數據。同時(shí)，在系統設計中，Master非常有利于橫向擴展，讓shuffle不會(huì )成為整個(gè)分布式系統的瓶頸。
讓New Shuffle模塊專(zhuān)注于shuffle，不依賴(lài)于外部計算模塊，從而計算模塊可以專(zhuān)注于計算，同時(shí)還避免了磁盤(pán)IO。然而New Shuffle帶來(lái)的問(wèn)題也隨之暴漏，其中影響比較重要的兩個(gè)就是：慢節點(diǎn)和數據重復。
慢節點(diǎn)。以shuffle寫(xiě)入過(guò)程中出現慢節點(diǎn)為例，通常包含兩個(gè)情況。首先，Shuffle自身慢節點(diǎn)，對比社區版本中只會(huì )影響到一個(gè)task，New Shuffle中常常會(huì )影響到一片集群。在這里，百度為每個(gè)Shuffle節點(diǎn)都配置了一個(gè)從節點(diǎn)，當Map檢測到一個(gè)慢節點(diǎn)時(shí)，系統會(huì )自動(dòng)切換到從節點(diǎn)。其次，DFS出現慢節點(diǎn)，這個(gè)情況下，Shuffle的從節點(diǎn)只能起到緩解作用。這種情況下，首先DFS系統會(huì )自動(dòng)檢測出慢節點(diǎn)，并進(jìn)行替換。比如，傳統的HDFS會(huì )以pipeline的形式進(jìn)行寫(xiě)入，而DFS則轉換為分發(fā)寫(xiě)。
在此之外，New Shuffle還需要解決更多問(wèn)題，比如資源共享和隔離等。同時(shí)，基于New Shuffle的機制，New Shuffle還面臨一些其他挑戰，比如Reduce全啟動(dòng)、數據過(guò)于分散、對DFS壓力過(guò)大、連接數等等。
數據重復。如上圖所示，這些問(wèn)題主要因為New Shuffle對上層組件缺少感知，這個(gè)問(wèn)題的解決主要使用task id和block id進(jìn)行去重。
New Shuffle展望
孫垚光表示，New Shuffle使用了通用的Writer和Reader接口，當下已經(jīng)支持百度MR和DCE（DAG、C++），同時(shí)即將對開(kāi)源Spark提供支持。在未來(lái)，New Shuffle無(wú)疑將成為更通用的組件，支持更多的計算模型。
百度美國硅谷研發(fā)中心高級架構師劉少山——Fast big data analytics with Spark on Tachyon
Tachyon是一個(gè)分布式的內存文件系統，可以在集群里以訪(fǎng)問(wèn)內存的速度來(lái)訪(fǎng)問(wèn)存在Tachyon里的文件。Tachyon是架構在分布式文件存儲和上層各種計算框架之間的中間件，主要負責將那些不需要落到DFS里的文件，落到分布式內存文件系統中，從而達到共享內存，以提高效率。1月10日下午的最后一場(chǎng)分享中，劉少山帶來(lái)了一場(chǎng)Tachyon的深入解析。
Tachyon和Spark
劉少山表示，在Spark使用過(guò)程中，用戶(hù)經(jīng)常困擾于3個(gè)問(wèn)題：首先，兩個(gè)Spark 實(shí)例通過(guò)存儲系統來(lái)共享數據，這個(gè)過(guò)程中對磁盤(pán)的操作會(huì )顯著(zhù)降低性能；其次，因為Spark崩潰所造成的數據丟失；最后，垃圾回收機制，如果兩個(gè)Spark實(shí)例需求同樣的數據，那么這個(gè)數據會(huì )被緩存兩次，從而造成很大的內存壓力，更降低性能。
使用Tachyon，存儲可以從Spark中分離處理，讓其更專(zhuān)注于計算，從而避免了上述的3個(gè)問(wèn)題。
Tachyon架構
劉少山從Spark的角度分享了Tachyon的部署。在與Spark搭配使用時(shí)，系統會(huì )建立一個(gè)Tachyon的job，通過(guò)Tachyon Client來(lái)訪(fǎng)問(wèn)同一個(gè)機器上的Tachyon Worker，也就是機器上的內存。而Tachyon Client則會(huì )與Tachyon Master交互，來(lái)清楚每個(gè)分節點(diǎn)所包含的數據。由此可見(jiàn)，在整個(gè)Tachyon 系統中，Master、Client和Worker為最重要的三個(gè)部分。
Tachyon Master。Master主要部件是Inode和Master Worker Info：Inode會(huì )負責系統的監視，Master Worker Info則存儲了所有Worker的信息。
Tachyon Worker。Worker主要負責存儲，其中Worker Storage是最主要的數據結構，包含Local data folder和Under File System兩個(gè)部分。其中Local data folder表示存在本地的Tachyon文件，Under File System則負責從HDFS中讀取Worker中未發(fā)現的數據。
Tachyon Client。Client為上層用戶(hù)提供了一個(gè)透明的機制，其TachyonFS接口負責數據請求。每個(gè)Client中有多個(gè)Tachyon File，其中Block In Stream負責文件讀取（Local Block In Stream負責本地機器讀取，Remote Block In Stream則負責讀取遠程機器）；Block Out Stream主要負責將文件寫(xiě)到本地機器上。在Client上，Master Client會(huì )與Master交互，Worker Client則與Client交互。
Tachyon在百度
為什么要使用Tachyon，劉少山指出，在百度，計算集群和存儲集群往往不在同一個(gè)地理位置的數據中心，在大數據分析時(shí)，遠程數據讀取將帶來(lái)非常高的延時(shí)，特別是ad-hoc查詢(xún)。因此，將Tachyon作為一個(gè)傳輸緩存層，百度通常會(huì )將之部署在計算集群上。首次查詢(xún)時(shí)，數據會(huì )從遠程存儲取出，而在以后的查詢(xún)中，數據就會(huì )從本地的Tacnyon上讀取，從而大幅的改善了延時(shí)。
在百度，Tachyon的部署還處于初始階段，大約部署了50臺機器，主要服務(wù)于ad-hoc查詢(xún)。
實(shí)踐中遭遇的挑戰
通過(guò)劉少山了解到，Tachyon的使用過(guò)程并不是一帆風(fēng)順，比如：因為T(mén)achyon需求對Block完全讀取，從而可能造成Blocks并未被緩存；有時(shí)候，雖然scheduler已經(jīng)確認了數據存在本地，Spark workers仍然從遠程blocks讀取，而緩存命中率也只有可憐的33%（如果你需要的是2號block，Tachyon會(huì )分別從1、2、3號block讀取，從而將block讀取了3份）。因此，劉少山表示，如果要使用好Spark與Tachyon，一定要對用例和Tachyon進(jìn)行充分的了解。
分享最后，劉少山還介紹了Hierarchical Storage Feature特性以及百度未來(lái)的工作，其中包括緩存替換策略等。
