Spark的基礎介紹和操作調優(yōu)

本篇內容介紹了“Spark的基礎介紹和操作調優(yōu)”的有關(guān)知識，在實(shí)際案例的操作過(guò)程中，不少人都會(huì )遇到這樣的困境，接下來(lái)就讓小編帶領(lǐng)大家學(xué)習一下如何處理這些情況吧！希望大家仔細閱讀，能夠學(xué)有所成！

Spark 基礎介紹

在討論spark調優(yōu)之前，先看看spark里的一些概念。

action

Action是得到非RDD結果的RDD操作。如Spark中有如下常見(jiàn)action操作： reduce, collect, count, first, take, takeSample, countByKey, saveAsTextFile

job

每個(gè)spark的action會(huì )被分解成一個(gè)job。

stage

一個(gè)job會(huì )被分成多組task，每組task稱(chēng)為一個(gè)stage。stage的劃分界限為以下兩種task之一：

• shuffleMapTask - 所有的wide transformation之前，可以簡(jiǎn)單認為是shuffle之前

• resultTask - 可以簡(jiǎn)單認為是take()之類(lèi)的操作

partition

RDD 包含固定數目的 partition， 每個(gè) partiton 包含若干的 record。

narrow tansformation （比如 map 和 filter）返回的 RDD，一個(gè) partition 中的 record 只需要從父 RDD 對應的 partition 中的 record 計算得到。同樣narrow transformation不會(huì )改變partition的個(gè)數。

task

被送到executor上執行的工作單元； 一個(gè)task只能做一個(gè)stage中的一個(gè)partition的數據。

操作調優(yōu)

• 調整在 stage 邊屆時(shí)的 partition 個(gè)數經(jīng)?？梢院艽蟪潭壬嫌绊懗绦虻膱绦行?

• associative reductive operation， 能使用reduceByKey時(shí)不使用groupByKey，因為grouByKey會(huì )把所有數據shuffle一遍，而reduceByKey只會(huì )Shuffle reduce的結果。

• 輸入和輸出結果不一樣時(shí)，不使用reduceByKey,而使用aggregateByKey;

aggregateByKey: Aggregate the values of each key, using given combine functions and a neutral "zero value". This function can return a different result type, U, than the type of the values in this RDD, V. Thus, we need one operation for merging a V into a U and one operation for merging two U's, as in scala.TraversableOnce. The former operation is used for merging values within a partition, and the latter is used for merging values between partitions. To avoid memory allocation, both of these functions are allowed to modify and return their first argument instead of creating a new U.

• 不要用flatMap-join-groupBy的模式，可以用cogroup;

• 當兩個(gè)reduceByKey的結果join時(shí)，如果大家的partition都一樣，則spark不會(huì )在join時(shí)做shuffle;

• 當一個(gè)內存能放得下的數據集join時(shí)，可以考慮broadcast而不使用join；

scala> val broadcastVar = sc.broadcast(Array(1, 2, 3))
broadcastVar: org.apache.spark.broadcast.Broadcast[Array[Int]] = Broadcast(0)

scala> broadcastVar.value
res0: Array[Int] = Array(1, 2, 3)

資源調優(yōu)

spark中的資源可以簡(jiǎn)單歸結為CPU和內存，而以下的參數會(huì )影響內存和CPU的使用。

• executor 越大并行性越好，越大每個(gè)executor所有的內存就越??；

• core，越大并行性越好;

HDFS client 在大量并發(fā)線(xiàn)程是時(shí)性能問(wèn)題。大概的估計是每個(gè) executor 中最多5個(gè)并行的 task 就可以占滿(mǎn)的寫(xiě)入帶寬。

• partition,如果比excutor*core小則很傻；越多每個(gè)partition占用的內存就越少；足夠大以后對性能提升不再有用。

我naive的認為應該這樣調整：

1. core = min(5,cpu核數)；

2. executor = instance數 * cpu核數 / core

3. 平均每instance的executor個(gè)數決定executor.memory,從而決定shuffle.memory和storage.memory;

4. 估計總數據量，即最大的shuffle時(shí)的數據大?。╯park driver運行記錄中會(huì )有shuffle size）;

5. 用4的結果除以3得到partition數，如果很小，把partition設成和(executor*core)的若干倍.