文 | Dong

來(lái)源 | 董的博客

概述
Apache Spark的高性能一定程度上取決于它采用的異步并發(fā)模型(這里指server/driver端采用的模型)，這與Hadoop 2.0(包括YARN和MapReduce)是一致的。Hadoop 2.0自己實(shí)現(xiàn)了類似Actor的異步并發(fā)模型，實(shí)現(xiàn)方式是epoll+狀態(tài)機(jī)，而Apache Spark則直接采用了開源軟件Akka，該軟件實(shí)現(xiàn)了Actor模型，性能非常高。盡管二者在server端采用了一致的并發(fā)模型，但在任務(wù)級(jí)別(特指Spark任務(wù)和MapReduce任務(wù))上卻采用了不同的并行機(jī)制： Hadoop MapReduce采用了多進(jìn)程模型，而Spark采用了多線程模型。
注意，本文的多進(jìn)程和多線程，指的是同一個(gè)節(jié)點(diǎn)上多個(gè)任務(wù)的運(yùn)行模式。無(wú)論是MapReduce和Spark，整體上看，都是多進(jìn)程：MapReduce應(yīng)用程序是由多個(gè)獨(dú)立的Task進(jìn)程組成的；Spark應(yīng)用程序的運(yùn)行環(huán)境是由多個(gè)獨(dú)立的Executor進(jìn)程構(gòu)建的臨時(shí)資源池構(gòu)成的。
多進(jìn)程模型便于細(xì)粒度控制每個(gè)任務(wù)占用的資源，但會(huì)消耗較多的啟動(dòng)時(shí)間，不適合運(yùn)行低延遲類型的作業(yè)，這是MapReduce廣為詬病的原因之一。而多線程模型則相反，該模型使得Spark很適合運(yùn)行低延遲類型的作業(yè)。
總之，Spark同節(jié)點(diǎn)上的任務(wù)以多線程的方式運(yùn)行在一個(gè)JVM進(jìn)程中，可帶來(lái)以下好處：
任務(wù)啟動(dòng)速度快，與之相反的是MapReduce Task進(jìn)程的慢啟動(dòng)速度，通常需要1s左右；
同節(jié)點(diǎn)上所有任務(wù)運(yùn)行在一個(gè)進(jìn)程中，有利于共享內(nèi)存。這非常適合內(nèi)存密集型任務(wù)，尤其對(duì)于那些需要加載大量詞典的應(yīng)用程序，可大大節(jié)省內(nèi)存。
同節(jié)點(diǎn)上所有任務(wù)可運(yùn)行在一個(gè)JVM進(jìn)程(Executor)中，且Executor所占資源可連續(xù)被多批任務(wù)使用，不會(huì)在運(yùn)行部分任務(wù)后釋放掉，這避免了每個(gè)任務(wù)重復(fù)申請(qǐng)資源帶來(lái)的時(shí)間開銷，對(duì)于任務(wù)數(shù)目非常多的應(yīng)用，可大大降低運(yùn)行時(shí)間。與之對(duì)比的是MapReduce中的Task：每個(gè)Task單獨(dú)申請(qǐng)資源，用完后馬上釋放，不能被其他任務(wù)重用，盡管1.0支持JVM重用在一定程度上彌補(bǔ)了該問題，但2.0尚未支持該功能。
盡管Spark的過線程模型帶來(lái)了很多好處，但同樣存在不足，主要有：
由于同節(jié)點(diǎn)上所有任務(wù)運(yùn)行在一個(gè)進(jìn)程中，因此，會(huì)出現(xiàn)嚴(yán)重的資源爭(zhēng)用，難以細(xì)粒度控制每個(gè)任務(wù)占用資源。與之相反的是MapReduce，它允許用戶單獨(dú)為Map Task和Reduce Task設(shè)置不同的資源，進(jìn)而細(xì)粒度控制任務(wù)占用資源量，有利于大作業(yè)的正常平穩(wěn)運(yùn)行。
下面簡(jiǎn)要介紹MapReduce的多進(jìn)程模型和Spark的多線程模型。
MapReduce多進(jìn)程模型

每個(gè)Task運(yùn)行在一個(gè)獨(dú)立的JVM進(jìn)程中；
可單獨(dú)為不同類型的Task設(shè)置不同的資源量，目前支持內(nèi)存和CPU兩種資源；
每個(gè)Task運(yùn)行完后，將釋放所占用的資源，這些資源不能被其他Task復(fù)用，即使是同一個(gè)作業(yè)相同類型的Task。也就是說(shuō)，每個(gè)Task都要經(jīng)歷“申請(qǐng)資源—> 運(yùn)行Task –> 釋放資源”的過程。
Spark多線程模型

每個(gè)節(jié)點(diǎn)上可以運(yùn)行一個(gè)或多個(gè)Executor服務(wù)；
每個(gè)Executor配有一定數(shù)量的slot，表示該Executor中可以同時(shí)運(yùn)行多少個(gè)ShuffleMapTask或者ResultTask；
每個(gè)Executor單獨(dú)運(yùn)行在一個(gè)JVM進(jìn)程中，每個(gè)Task則是運(yùn)行在Executor中的一個(gè)線程；
同一個(gè)Executor內(nèi)部的Task可共享內(nèi)存，比如通過函數(shù)SparkContext#broadcast廣播的文件或者數(shù)據(jù)結(jié)構(gòu)只會(huì)在每個(gè)Executor中加載一次，而不會(huì)像MapReduce那樣，每個(gè)Task加載一次；
Executor一旦啟動(dòng)后，將一直運(yùn)行，且它的資源可以一直被Task復(fù)用，直到Spark程序運(yùn)行完成后才釋放退出。
總結(jié)
總體上看，Spark采用的是經(jīng)典的scheduler/workers模式，每個(gè)Spark應(yīng)用程序運(yùn)行的第一步是構(gòu)建一個(gè)可重用的資源池，然后在這個(gè)資源池里運(yùn)行所有的ShuffleMapTask和ResultTask(注意，盡管Spark編程方式十分靈活，不再局限于編寫Mapper和Reducer，但是在Spark引擎內(nèi)部只用兩類Task便可表示出一個(gè)復(fù)雜的應(yīng)用程序，即ShuffleMapTask和ResultTask)，而MapReduce應(yīng)用程序則不同，它不會(huì)構(gòu)建一個(gè)可重用的資源池，而是讓每個(gè)Task動(dòng)態(tài)申請(qǐng)資源，且運(yùn)行完后馬上釋放資源。
end