19網(wǎng)絡(luò)視角如何有效降低開銷

這么多存儲格式和外部存儲系統(tǒng)交叉在一起又會有無數(shù)種組合，并且每一種組合都有它的應(yīng)用場景。那么，我們該怎么判斷網(wǎng)絡(luò)開銷會出現(xiàn)在哪些場景下呢？其實，，它一共有4種：PROCESS_LOCALJVM進程中NODE_LOCAL：JVM 任務(wù)與數(shù)據(jù)是跨機架、甚至是跨DC（DataCenter，數(shù)據(jù)中心）的關(guān)系根據(jù)定義我們很容易判斷出，不同本地性級別下的計算任務(wù)是否會引入磁盤或網(wǎng)絡(luò)開銷，結(jié)果如下表所示。從表格中我們不難發(fā)現(xiàn)，從PROCESS_LOCAL到AY是逐級變差的。在讀取數(shù)據(jù)源階段，數(shù)據(jù)還未加載到內(nèi)存，任務(wù)沒有辦法調(diào)度到PROCESS_LOCAL級別。因此，這個階段我們能夠調(diào)度的最佳級別是NODE_LOCAL。根據(jù)NODE_LOCAL的定義，在這個級別下，調(diào)度的目標節(jié)點至少在磁盤上存有Spark計算任務(wù)所需的數(shù)據(jù)分片。這也就意味著，在集群部署上，Spark集群與外部存儲系統(tǒng)在物理上是緊緊耦合在一起的。相反，如果Spark集群與存儲集群在物理上是分開的，那么任務(wù)的本地性級別只能退化到RACK_LOCAL，甚至是ANY，來通過網(wǎng)絡(luò)獲取所需的數(shù)據(jù)分片。因此，對于Spark加HDFS和Spark加MongoDB決于它們的部署模式。物理上緊耦合，在NODE_LOCAL級別下，Spark用磁盤I/O替代免網(wǎng)絡(luò)開銷。值得一提的是，在企業(yè)的私有化C合的方式來提升數(shù)據(jù)訪問效率。但是在公有云環(huán)境中，計算集群在物理上往往和存儲系統(tǒng)隔離，因此數(shù)據(jù)源的讀取只能走網(wǎng)絡(luò)。上做規(guī)劃，從而在最開始部署Spark集群的時候就提前做好準備。Shuffle。Shuffle法，就是把ShuffleJoins轉(zhuǎn)化為Broadast變量那幾講有過詳細的講解，你可以翻回去看一看。盡管廣播變量的創(chuàng)建過程也會引入網(wǎng)絡(luò)傳輸，但是，兩害相權(quán)取其輕，相比Shuffle的網(wǎng)絡(luò)開銷，廣播變量的開銷算是小巫見大巫了。遵循“能省則省”的原則，把Shuffle消除掉自然是最好的。如果實在沒法避免Shuffle，我們要盡可能地在計算中多使用Map端聚合，去減少需要在網(wǎng)絡(luò)中分發(fā)的數(shù)據(jù)量。這方面的典型做法就是用reduceByKey、aggregateByKey替換groupByKey，不過在RDDAPI使用頻率越來越低的當下，這個調(diào)優(yōu)技巧實際上早就名存實亡了。但是，Map端聚合的在絕大多數(shù)2C（ToConsumer）的業(yè)務(wù)場景中，我們都需要刻畫用戶畫像。我們的小例表內(nèi)容唯一，也就是不存在重復(fù)的興趣項，用戶表的Schema如下表所示。要獲取群組興趣列表，我們應(yīng)該先按照groupd然后再把列表中的興趣項展平，最后去重得到內(nèi)容唯一的興趣列表。應(yīng)該說思路還是蠻簡單的，我們先來看第一版實現(xiàn)代碼。valvalfilePath:String=valdf=spark.read.parquent(filePath)這版實現(xiàn)分別用collect_list、flatten和array_distinct，來做興趣列表的收集、展平和去重操作，它完全符合業(yè)務(wù)邏輯。不過，見到“收集”類的操作，比如groupByKey，以及這里的collect_list，我們應(yīng)該本能地提高警惕。因為這類操作會把最細粒度的全量數(shù)據(jù)在那我們是不是可以把它們提前到Map端，從而減少Shuffle中需要分發(fā)的數(shù)據(jù)量呢？當然Map端做一次去重，然后去查找DataFrame開發(fā)API，看看有沒有與collect_list對應(yīng)的Map端聚因此，在數(shù)據(jù)處理環(huán)節(jié)，我們要遵循“能省則省”的開發(fā)原則，主動削減計算過程中的網(wǎng)eJoinsrtJoinshhMp除了Shuffle之外，還有個操作也會讓數(shù)據(jù)在網(wǎng)絡(luò)中分發(fā)，這個操作很隱蔽，我們經(jīng)常注意不到它，它就是多副本的RDD緩存。比如說，在實時流處理這樣的場景下，對于系統(tǒng)的高可用性，應(yīng)用的要求比較高，因此你可能會用“_”甚至是“_3”的存儲模式，在內(nèi)存和磁盤中緩存多份數(shù)據(jù)拷貝。當數(shù)據(jù)副本數(shù)大于1雖然這看上去只是存儲模式字符串的一個微小改動，但在運行時，它會帶來很多意想不到的開銷。因此，如果你的應(yīng)用對高可用特性沒有嚴格要求，我建議你盡量不要濫用多副本的RDD緩存，最后就到了數(shù)據(jù)傳輸?shù)沫h(huán)節(jié)。我們知道，在落盤或是在網(wǎng)絡(luò)傳輸之前，數(shù)據(jù)都是需要先進行序列化的。在Spk中，有兩種序列化器供開發(fā)者選擇，分別是Jaaserializer和KryoSerializer。park官方和網(wǎng)上的技術(shù)博客都會推薦你使用KryoSeralizer來提高效率，通常來說，KryoSeriaizer相比Javaserializer，在處理效率和存儲效率兩個方面都會勝出數(shù)倍。因此，在數(shù)據(jù)分發(fā)之前，使用KryoSerializer開銷。不過，經(jīng)常有同學(xué)向我抱怨：“為什么我用了KryoSerializer，序列化之后的數(shù)據(jù)尺寸反而比Javaserializeror列化的工作，但它序列化的每一條數(shù)據(jù)記錄都會帶一個類名字，這個類名字是通過反射機那該怎么向KryoSerializer注冊自定義類型呢？其實非常簡單，我們只需要在SparkConfregisterKryoClasses，代碼示例如下所示。////向KryoSerializervalconf=newSparkConf().setMaster(“”).setAppName(“”)classOf[HashMap[String,String]],另外，與KryoSerializer有關(guān)的配置項，我也把它們匯總到了下面的表格中，方便你隨時查找。其中，spark.serializer可以明確指定Spark采用KryoSeriaizer序列化器。而spark.kryo.registrationRequired就比較有意思了，如果我們把它設(shè)置為True，當KryoSeralizer出異常，并且中斷任務(wù)執(zhí)行。這么做的好處在于，在開發(fā)和調(diào)試階段，它能幫我們捕捉那些忘記注冊的類型。JavaserializerKryoSerializerJavaSerializer只要你出具家庭住址，JavaSeralizer里都有哪些物件，他們對不同種類的物品有一套自己的打包標準，可以幫你省去很多麻煩。不過，JavaSerializerKryoSerializer屬于市場新貴，在打包速度和包裹尺寸方面都遠勝JavaSerializer。KryoSerializer會以最緊湊的方式打包，一寸空間也不浪費，因此，所有包裹用一輛小貨車就能裝下。但是，訂閱KryoSerializer的托管服務(wù)之前，用戶需要提供詳盡的物品明細表，因此，很多用戶都嫌麻煩，KryoSerializer市場占有率也就一直上不去。好啦，現(xiàn)在你是不是對JavaSerializer和KryoSerializer有了更深的理解了？由此可見，如果你想要不遺余力去削減數(shù)據(jù)傳輸過程中的網(wǎng)絡(luò)開銷，就可以嘗試使用KryoSerializer來做數(shù)據(jù)的序列化。相反，要是你覺得開發(fā)成本才是核心痛點，那采用默認的JavaSerializer首先，在數(shù)據(jù)讀取階段，要想獲得NODE_LOCAL的本地級別，我們得讓Spark集群與外部存儲系統(tǒng)在物理上緊緊耦合在一起。這樣，Spark就可以用磁盤I/O替代網(wǎng)絡(luò)開銷獲取數(shù)據(jù)，否則，本地級別就會退化到RACK_LOCAL或者ANY，那網(wǎng)絡(luò)開銷就無法避免。BroadcastJoins來避免Shuffle引入的網(wǎng)絡(luò)開銷。如果確實沒法避免Shuffle，我們可以在計算中多使用Map端聚合，減少需要在網(wǎng)絡(luò)中分發(fā)的數(shù)據(jù)量。另外，如果應(yīng)用對于高可用的要求不高，那我們應(yīng)該盡量避免副本數(shù)量大于1的存儲模式，避免副本跨節(jié)點拷貝帶來KryoSerializerKryoSerializerMapcollect_listMap你還能想到哪些Map 18|磁盤視角：如果內(nèi)存無限大，磁盤還有用武之地嗎 20|RDD和DataFrame：既生瑜、何生的stage生成的task都是需要進行shuffle，最起碼也是racknode?！詈玫臅r候，就是PROCESS_LOCAL。其實，Alluxio的文件存儲，本質(zhì)上相當于OFFHEAP的ca因此，很顯然，Cache是可以保證PROCESS_LOCAL的。但是，要做到PROCESS_LOCAL，卻不是一定非要做cache，BroadcastJoin機制下，小表的廣播，讀取的時候也是PROCESS_LOCALocatedjoin，你不妨搜搜這個數(shù)據(jù)關(guān)聯(lián)的計算過程，它的計算，也是PROCESS_LOCAL的。總之一句話，CachePROCESS_LOCAL，但是要做到PROCESS_LOCAL，不一定非要做Cache。SparkSQL哈~SparkSQL內(nèi)置了很多優(yōu)化機制（Catalyst、Tungsten），這個DataFrame要想實現(xiàn)map端預(yù)聚合只能靠優(yōu)化器自己了吧比較抽象層次比較高了，靈活deflocalDistinct(set:Set[String],b:String):mutable.Set[String]=可兼得么，lowlevelAPI比如RDD，高階算子，開發(fā)靈活性非常高，但是優(yōu)化空間有限；highlevelAPI比如DataFrame，SparkSQL優(yōu)化空間大，但是就沒有RDD開發(fā)起來那么靈活了。reduceBykeymap聚合不影響正確性的都可以先在map端做聚合executornodelocal，再次就是同個機架下，實現(xiàn)racklocal。數(shù)據(jù)Shuffle時，可以考慮使用Broadcast代… 19|deflocalDistinct(set:Set[String],b:String):mutable.Set[String]=展開作者回復(fù):Nice~aggregateByKey玩兒的很溜啊~不過咱們這個例子其實可以用collect_set哈可兼得么，lowlevelAPI比如RDD，高階算子，開發(fā)靈活性非常高，但