共計 2403 個字符，預計需要花費 7 分鐘才能閱讀完成。
本篇文章給大家分享的是有關怎么給 Spark 傳遞函數(shù)，丸趣 TV 小編覺得挺實用的，因此分享給大家學習，希望大家閱讀完這篇文章后可以有所收獲，話不多說，跟著丸趣 TV 小編一起來看看吧。
相信很多人在開始用 Spark 的時候一定會遇到 Task not serializable 的問題，這種問題大多數(shù)都是在 RDD 的算子中調(diào)用了不能序列化的對象引起的。為什么傳入算子中的對象一定要能夠序列化呢？這就要從 Spark 本身說起，Spark 是一個分布式的計算框架，RDD(Resilient Distributed Datasets, 彈性分布式數(shù)據(jù)集）是對分布式數(shù)據(jù)集的抽象，數(shù)據(jù)實際上是分布在集群的各個節(jié)點的，通過 RDD 進行抽象，讓用戶感覺好像是在本地交互一樣。但是實際的運算中，算子中的操作都要發(fā)送到計算節(jié)點（Executor）端來執(zhí)行，這就要求傳入算子中的對象可以進行序列化。
Spark 的算子很大程度上是上通過向集群上的驅(qū)動程序傳遞函數(shù)來實現(xiàn)的，編寫 Spark 應用的關鍵就是使用算子（或者稱為轉(zhuǎn)換），給 Spark 傳遞函數(shù)來實現(xiàn)。常用的向 Spark 傳遞函數(shù)的方式有兩種（來自于 Spark 官方文檔，Spark 編程指南）：
第一種：匿名函數(shù)，處理的代碼比較少的時候，可以采用匿名函數(shù)，直接寫在算子里面：
myrdd.map(x = x+ 1)
第二種：全局單例對象中的靜態(tài)方法：先定義 object 對象 MyFunctions，以及靜態(tài)方法：funcOne, 然后傳遞 MyFunctions.funcOne 給 RDD 算子。
object MyFunctions { def funcOne(s: String): String = { ... }
}
myRdd.map(MyFunctions.funcOne)
在業(yè)務員開發(fā)中，需要把 RDD 的引用傳遞給某一個類的實例的某個方法，傳遞給 RDD 的函數(shù)，為類實例的實例方法：
class MyClass { def funcOne(s: String): String = { ... }
def doStuff(rdd: RDD[String]): RDD[String] = { rdd.map(funcOne }
}
在這個例子中，我們定義了一個類 MyClass，類的實例方法 doStuff 中傳入了一個 RDD，RDD 算子中調(diào)用了類的另外一個實例方法 funcOne, 在我么 New 一個 MyClass 的實例并調(diào)用 doStuff 的方法的時候，需要講整個實例對象發(fā)給集群，所以類 MyClass 必須可以序列化，需要 extends Serializable。
相似的，訪問方法外部的對象變量也會引用整個對象，需要把整個對象發(fā)送到集群：
class MyClass {
val field = Hello
def doStuff(rdd: RDD[String]): RDD[String] = { rdd.map(x = field
+ x) }}
為了避免整個對象都發(fā)送給集群，可以定義一個局部變量來保存外部對象 field 的引用，這種情況尤其在一些大對象里，可以避免整個對象發(fā)送到集群，提高效率。
def doStuff(rdd: RDD[String]): RDD[String] = {
val field_ = this.field
rdd.map(x = field_ + x)
}
Spark 應用最終是要在集群中運行的，許多問題在單一的本地環(huán)境中無法暴露出來，有時候經(jīng)常會遇到本地運行結果和集群運行結果不一致的問題，這就要求開發(fā)的時候多使用函數(shù)式編程風格，盡量使的寫的函數(shù)都為純函數(shù)。純函數(shù)的好處是：無狀態(tài), 線程安全, 不需要線程同步，應用程序或者運行環(huán)境 (Runtime) 可以對純函數(shù)的運算結果進行緩存, 運算加快速度。
那么什么是純函數(shù)了？
純函數(shù)（Pure Function）是這樣一種函數(shù)——輸入輸出數(shù)據(jù)流全是顯式（Explicit）的。顯式（Explicit）的意思是，函數(shù)與外界交換數(shù)據(jù)只有一個唯一渠道——參數(shù)和返回值；函數(shù)從函數(shù)外部接受的所有輸入信息都通過參數(shù)傳遞到該函數(shù)內(nèi)部；函數(shù)輸出到函數(shù)外部的所有信息都通過返回值傳遞到該函數(shù)外部。如果一個函數(shù)通過隱式（Implicit）方式，從外界獲取數(shù)據(jù)，或者向外部輸出數(shù)據(jù)，那么，該函數(shù)就不是純函數(shù)，叫作非純函數(shù)（Impure Function）。隱式（Implicit）的意思是，函數(shù)通過參數(shù)和返回值以外的渠道，和外界進行數(shù)據(jù)交換。比如，讀取全局變量，修改全局變量，都叫作以隱式的方式和外界進行數(shù)據(jù)交換；比如，利用 I /O API（輸入輸出系統(tǒng)函數(shù)庫）讀取配置文件，或者輸出到文件，打印到屏幕，都叫做隱式的方式和外界進行數(shù)據(jù)交換。
在計算過程中涉及到對象的交互時，盡量選用無狀態(tài)的對象，比如對于一個 bean，成員變量都為 val 的，在需要數(shù)據(jù)交互的地方 new 一個新的。
關于（commutative and associative）交換律和結合律。在傳遞給 reudce,reduceByKey, 以及其他的一些 merge，聚合的操作中的函數(shù)必須要滿足交換律和結合律，交換律和結合律就是我們數(shù)學上學過的：
a + b = b + a，a + b + c = a + (b + c)
定義的函數(shù) func(a,b)和 f(b,a)應該得到相同的結果，f(f(a,b),c)和 f(a,f(b,c))應該得到相同的結果。
最后說一下廣播變量和累加器的使用。在程序中不要定義一個全局的變量，如果需要在多個節(jié)點共享一個數(shù)據(jù)，可以采用廣播變量的方法。如果需要一些全局的聚合計算，可以使用累加器。
以上就是怎么給 Spark 傳遞函數(shù)，丸趣 TV 小編相信有部分知識點可能是我們?nèi)粘９ぷ鲿姷交蛴玫降摹ＯＭ隳芡ㄟ^這篇文章學到更多知識。更多詳情敬請關注丸趣 TV 行業(yè)資訊頻道。
