1月24日，一場基于Spark和Redis組成的分布式系統(tǒng)實踐分享由Spark資深布道者陳超和豌豆莢資深系統(tǒng)架構(gòu)師劉奇聯(lián)手打造,。

陳超：Spark Ecosystem & Internals

陳超（@CrazyJvm）,，Spark布道者

在分享中，陳超首先簡短的介紹了Spark社區(qū)在2014年的發(fā)展：目前Spark的發(fā)布版本是1.2,，整個2014年Spark共發(fā)布了3個主要版本——1.0,、1.1、1.2,。隨后,，陳超對Spark生態(tài)圈進行了詳細的分析：

Spark：What & Why？

Spark是一個非常快,，并且普適性非常強的一個大數(shù)據(jù)處理引擎,。談到Spark，首先就是一些常見特性：速度快,、易用,、通用和兼容Hadoop。首先通用,，Spark可以支撐批處理,、流計算、圖計算,、機器學習等眾多應用場景,；其次，與Hadoop良好的兼容,。鑒于大多數(shù)的企業(yè)仍選用HDFS來存數(shù)據(jù),，Spark的設(shè)計與HDFS有著非常好的兼容性——假如數(shù)據(jù)存儲在HDFS，那么不做任何數(shù)據(jù)遷移工作就可以直接使用Spark,。

Spark vs. Hadoop

對于為什么要選擇Spark,，如上圖所示，陳超從迭代計算和HDFS同批數(shù)據(jù)的多維度查詢兩個方面將之與Hadoop進行了對比：

• 迭代計算,。在這個場景下,，Hadoop需要多次讀寫HDFS（磁盤），造成了大量的IO和序列化,、反序列化等額外開銷,。此外，每次寫HDFS都需要寫入3份,，因此造成了備份方面的開銷,。
• HDFS同批數(shù)據(jù)的多維度查詢。對HDFS同一批數(shù)據(jù)做成百或上千維度查詢時,，Hadoop每次做一個獨立的query,，也就是每次都要從磁盤讀取這個數(shù)據(jù)。因為每次都從磁盤中讀取同一組數(shù)據(jù),，效率顯然可以繼續(xù)提高,。

而在這兩種場景中，Spark可以使用內(nèi)存緩存中間/常用數(shù)據(jù),，從而在避免磁盤IO開銷的同時，還將大幅度提高性能,。

Why Spark is so Fast,？

Spark一直以快速著稱，那么除下之前所說的內(nèi)存，又是什么特性讓Spark可以如此之快,？在這里,，陳超提到了DAG（有向無環(huán)圖，下文詳細介紹）,、Thread Model（線程模型）和Optimization（比如延遲調(diào)度）3個方面,。

Thread Model。Hadoop基于進程模型,，每次啟動一個task都需要新啟動一個子JVM進行計算,，可能也會存在JVM Reuse，這里即使避開JVM Reuse中存在的問題不談,，每次JVM啟動時已經(jīng)造成了不菲的開銷,。而Spark在應用程序啟動時就啟動了線程池，所以任務的啟動開銷非常小,。

Optimization——延遲調(diào)度,。當任務下達到某臺主機時，恰好該主機的計算資源（CPU,、內(nèi)存等）已被耗盡,，這個時候，Spark會采用延遲調(diào)度的機制,，讓其等待一小會,，而不是將該臺主機上需要計算的數(shù)據(jù)通過網(wǎng)絡(luò)傳輸?shù)搅硗獾闹鳈C上。使用這個機制,，在計算數(shù)據(jù)體積非常大時,，有著很大的優(yōu)勢。 也就是所謂的“讓計算跟著數(shù)據(jù)走,，而不是數(shù)據(jù)跟著計算走”,。

Spark解析

伯克利數(shù)據(jù)分析協(xié)議棧

其中包括：資源管理框架，Apache YARN,、Apache Mesos,；基于內(nèi)存的分布式文件系統(tǒng)，Tachyon,；隨后是Spark,，更上面則是實現(xiàn)各種功能的系統(tǒng)，比如機器學習MLlib庫,，圖計算GraphX,，流計算Spark Streaming。再上面比如：SparkR,，分析師的最愛,；BlinkDB,，我們可以強迫它幾秒鐘內(nèi)給我們查詢結(jié)果。

正是這個生態(tài)圈,，讓Spark可以實現(xiàn)“one stack to rule them all”,，它既可以完成批處理也可以從事流計算，從而避免了去實現(xiàn)兩份邏輯代碼,。而整個Spark的理論基礎(chǔ)就是RDD：

RDD的核心理念

RDD可以想象為一個個的partitions,，退一步也可理解為一個非常大的List(1,2,....9)，使用3個Partion分別保存這個List的3個元素,，而每個partition（或者split）都會有一個函數(shù)去計算,。同時，RDD之間是可以相互依賴的,。然后,，可以為Key-value RDD指定partitioner, RDD中的每個split也都有各自的preferred location。

最后一個preferred locations,，這個理念存在于當下的眾多分布式系統(tǒng)中,，也就是計算跟著數(shù)據(jù)走。通常情況下,，轉(zhuǎn)移計算的時間遠遠小于轉(zhuǎn)移數(shù)據(jù)的時間,。對于Hadoop來說，因為數(shù)據(jù)在磁盤中,，磁盤本地性通常達到了頂峰,，而對于Spark來講，因為數(shù)據(jù)（可以）保存在內(nèi)存中,，所以內(nèi)存本地性才具備最高優(yōu)先級,。

運行原理

上圖表述了Spark運行原理：rdd1、rdd2,、rdd3等等一直轉(zhuǎn)換到另外一個RDD,。需要注意的是，這里面存在的是一個延遲的執(zhí)行,，也就是轉(zhuǎn)換不會立刻執(zhí)行,。Spark只會在元數(shù)據(jù)中記錄這個過程，但是不會真正的執(zhí)行,，這個要注意一點,，只有在碰到action的時候才會真正的去執(zhí)行。這個時候需要注意的是,，比如上圖RDD2所做的cache,，這個操作同樣是lazy的，同樣在碰到action的時候才會執(zhí)行,。就在這里,，坑出現(xiàn)了,，即使persist與cache使用的是相同的接口，但是unpersist卻是eager的,。從1.1版本開始，cache其實已經(jīng)有了更安全的做法,，但是涉及過多內(nèi)核細節(jié),，這里就不做多的解釋。

RDD的依賴性

narrow dependency和wide dependency是Spark中另外兩個重要的概念,。對比后者,，narrow dependency無論是在從容錯上，還是在執(zhí)行效率上都占有優(yōu)勢,。

ClusterManager：目前來講,，在國內(nèi)采用率更大的顯然是YARN。

Cluster overview

Sparkcontext,，寫代碼時生成,，并向ClusterManager請求資源。ClusterManager會負責連接到Worker Node取得資源,，其中executor才是task的真正執(zhí)行者,。這里有三個需要注意的點：第一，ClusterManager是可插拔的,，可以任意選擇,；第二點，因為driver program需求發(fā)送任務給Worker Node,，因此提交任何的地方不要離Worker Node特別遠,。第三點比較重要的一點，每個應用程序在每個Worker Node上都會有獨立的executor,，并且不同應用程序的executor(間)是不可以共享數(shù)據(jù)的,。

PS：YARN通過Container來封裝資源，因此在YARN中Worker對應的是Container,。

調(diào)度

最初,，Spark程序會隱式地建立一個邏輯上有向無環(huán)圖（DAG），隨后DAGScheduler會將DAG切分成一個個stage,，隨后這些stage會被傳送給TaskSchedluer,，之后再傳送給Worker上的excutor執(zhí)行。其中excutor會以多線程的模式執(zhí)行,。

Shuffle

從理論上講,，Spark Shuffle從未超過MapReduce，直到改完以后才OK,。當下,，Shuffle使用的是基于PULL的模式,，中間文件會寫到磁盤，同時,，在每個partition都會建立hash map,。需要注意的是，在可以跨keys spill的同時,，主機內(nèi)存必須可以裝進單key-value,。

在監(jiān)控上，之前的版本中,，只有當一個任務結(jié)束時,，才可以收集這個任務的運行數(shù)據(jù)，這點在當下的版本已被改進,。

生態(tài)系統(tǒng)簡析

Spark Streaming：Spark Streaming實質(zhì)上仍然是批處理,，但是把之前大的批處理拆為小的batch。同時,，當下Spark Streaming已支持限流,，當流量很大時，Spark可以擋住,。此外,，它還可以支持實時機器學習。在Spark Streaming中,，數(shù)據(jù)丟失一般因為兩種情況——worker failure和driver failure,。在之前版本中，可能會存在小部分的數(shù)據(jù)丟失,，而在1.2版本發(fā)布后,，reliable receiver模式保證了所有數(shù)據(jù)不會丟失，這點在Kafka的連接上非常適用,。

MLlib：當下的算法已經(jīng)非常豐富,，包括分類、聚類,、回歸,、協(xié)同過濾、降維等等,。ML Pipeline可以大幅度的減少開發(fā)時間,，它可以幫開發(fā)者打通數(shù)據(jù)收集、數(shù)據(jù)清理,、特征提取,，模型訓練，測試,、評估,、上線整個流程,。

Graphx：在這里，Spark的優(yōu)勢是既能處理表視圖,，也能處理圖視圖,。

Spark SQL：Spark生態(tài)圈中最火的組件，目的很簡單,，用來支持SQL標準,。對比Spark SQL，因為基于MapReduce的進程模型,，Hive中存在許多一直未修復的多線程bug。值得一提的是,，Spark SQL的貢獻者中,，一半以上是華人。

Tachyon可以支撐幾乎所有框架

Tachyon：內(nèi)存分布式系統(tǒng),，讓不同的Job或者框架分享數(shù)據(jù),，從而繞過HDFS，以更快地速度執(zhí)行,。同時,，它還可以避免任務失敗時的數(shù)據(jù)重算。最后,，Tachyon可以讓系統(tǒng)避免多次GC,。

SparkR：讓R語言調(diào)用Spark。原理是Spark Context通過JNI調(diào)用Java Spark Context,，隨后通過Worker上的Excutor調(diào)用R的shell來執(zhí)行?，F(xiàn)在存在的問題是，每次task執(zhí)行時都需要啟動R shell,，所以還亟待優(yōu)化,。

BlinkDB，一個任性的數(shù)據(jù)庫

BlinkDB：很任性的一個數(shù)據(jù)庫,，允許操作者帶著time bounds或者error bounds去查,。原理是在原始數(shù)據(jù)上維護一組多維樣本，當然其中還需要一個動態(tài)的樣本選擇策略,。

JobServer：提供了一個RESTful接口來提交和管理Apache Spark job,、jars及job contexts，即Spark as a Service,。

劉奇：Codis Design & Implementation

劉奇（@goroutine）,，豌豆莢資深系統(tǒng)架構(gòu)師

在劉奇的分享中，他首先介紹了Redis在豌豆莢的使用歷程——單實例==》多實例,，業(yè)務代碼中做sharding==》單個Twemproxy==》多個Twemproxy==》Codis,，豌豆莢自己開發(fā)的分布式Redis服務,。在大規(guī)模的Redis使用過程中，他們發(fā)現(xiàn)Redis受限于多個方面：單機內(nèi)存有限,、帶寬壓力,、單點問題、不能動態(tài)擴容以及磁盤損壞時的數(shù)據(jù)搶救,。

通過劉奇我們了解到,，Redis通常有3個使用途徑：客戶端靜態(tài)分片，一致性哈希,；通過Proxy分片,，即Twemproxy；還有就是官方的Redis Cluster,，但至今無一個新版本,。隨后劉奇更詳細的分析了為什么不使用Twemproxy和Redis Cluster：

Twemproxy：最大的痛點是無法平滑的擴容或者縮容，甚至修改配置都需要重啟服務,；其次,，不可運維，甚至沒有Dashboard,。

Redis Cluster（官方）：無中心化設(shè)計,，程序難以編寫；代碼有點嚇人,，clusterProcessPacket函數(shù)有426行,，人腦難以處理所有的狀態(tài)切換；遲遲沒有正式版本,，等了4年之久,；目前還缺乏最佳實踐，沒有人編寫Redis Cluster的若干條注意事項,；整個系統(tǒng)高度耦合,，升級困難。

劉奇表示,，雖然我們有眾多的選擇,，比如Tair、Couchbase等,，但是如果你需要更復雜和優(yōu)秀的數(shù)據(jù)結(jié)構(gòu),，Redis可稱為不二之選?；谶@個原因,，在Redis之上，豌豆莢設(shè)計了Codis，并將之開源,。

Codis

既然重新設(shè)計,，那么Codis首先必須滿足自動擴容和縮容的需求，其次則是必須避免單點故障和單點帶寬不足,，做一個高可用的系統(tǒng),。在這之后，基于原有的遺留系統(tǒng),，還必須可以輕松地將數(shù)據(jù)從Twemproxy遷移到Codis,，并實現(xiàn)良好的運維和監(jiān)控?；谶@些,，Codis的設(shè)計躍然紙面：

然而，一個新系統(tǒng)的開發(fā)并不是件容易的事情,，特別是一個復雜的分布式系統(tǒng),。劉奇表示，雖然當時團隊只有3個人,，但是他們幾乎考量了可以考量的各種細節(jié)：

• 盡量拆分，簡化每個模塊,，同時易于升級
• 每個組件只負責自己的事情
• Redis只作為存儲引擎
• Proxy的狀態(tài)
• Redis故障判定是否放到外部,，因為分布式系統(tǒng)存活的判定異常復雜
• 提供API讓外部調(diào)用，當Redis Master丟失時,，提升Slave為Master
• 圖形化監(jiān)控一切：slot狀態(tài),、Proxy狀態(tài)、group狀態(tài),、lock,、action等等

而在考量了一切事情后，另一個爭論擺在了眼前——Proxy或者是Smart Client：Proxy擁有更好的監(jiān)控和控制,，同時其后端信息亦不易暴露,，易于升級；而Smart Client擁有更好的性能,，及更低的延時,，但是升級起來卻比較麻煩。對比種種優(yōu)劣,，他們最終選擇了Proxy,，無獨有偶，在codis開源后,，twitter的一個分享提到他們也是基于proxy的設(shè)計,。

Codis主要包含Codis Proxy（codis-proxy）、Codis Manager（codis-config）,、Codis Redis（codis-server）和ZooKeeper四大組件,，每個部分都可動態(tài)擴容,。

codis-proxy 。客戶端連接的Redis代理服務,，本身實現(xiàn)了Redis協(xié)議,，表現(xiàn)很像原生的Redis （就像 Twemproxy）。一個業(yè)務可以部署多個 codis-proxy,，其本身是無狀態(tài)的,。

codis-config。Codis 的管理工具,，支持添加/刪除Redis節(jié)點,、添加/刪除Proxy節(jié)點、發(fā)起數(shù)據(jù)遷移等操作,。codis-config自帶了一個http server,，會啟動一個dashboard，用戶可以在瀏覽器上觀察 Codis 集群的運行狀態(tài),。

codis-server,。Codis 項目維護的一個Redis分支，加入了slot的支持和原子的數(shù)據(jù)遷移指令,。

ZooKeeper,。Codis依賴ZooKeeper來存放數(shù)據(jù)路由表和codis-proxy節(jié)點的元信息，codis-config發(fā)起的命令會通過 ZooKeeper同步到各個存活的codis-proxy,。

最后,，劉奇還介紹詳細的了Codis中Migration、lock(rwlock)等操作的實現(xiàn)過程和原理,，以及從Twemproxy遷移到Codis的詳細操作,。更多Codis詳情可移步Clodis開源頁GitHub。