怎么給Spark傳遞函數(shù)

本篇文章給大家分享的是有關(guān)怎么給Spark傳遞函數(shù)，小編覺得挺實用的，因此分享給大家學(xué)習(xí)，希望大家閱讀完這篇文章后可以有所收獲，話不多說，跟著小編一起來看看吧。

 相信很多人在開始用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應(yīng)用的關(guān)鍵就是使用算子（或者稱為轉(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è)務(wù)員開發(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應(yīng)用最終是要在集群中運行的，許多問題在單一的本地環(huán)境中無法暴露出來，有時候經(jīng)常會遇到本地運行結(jié)果和集群運行結(jié)果不一致的問題，這就要求開發(fā)的時候多使用函數(shù)式編程風(fēng)格，盡量使的寫的函數(shù)都為純函數(shù)。純函數(shù)的好處是：無狀態(tài),線程安全,不需要線程同步，應(yīng)用程序或者運行環(huán)境(Runtime)可以對純函數(shù)的運算結(jié)果進行緩存,運算加快速度。

    那么什么是純函數(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 一個新的。

       關(guān)于（commutative and associative）交換律和結(jié)合律。在傳遞給reudce,reduceByKey,以及其他的一些merge，聚合的操作中的函數(shù)必須要滿足交換律和結(jié)合律，交換律和結(jié)合律就是我們數(shù)學(xué)上學(xué)過的：

      a + b = b + a，a + b + c =  a + (b + c)

定義的函數(shù)func(a,b)和f(b,a)應(yīng)該得到相同的結(jié)果，f(f(a,b),c)和f(a,f(b,c))應(yīng)該得到相同的結(jié)果。

     最后說一下廣播變量和累加器的使用。在程序中不要定義一個全局的變量，如果需要在多個節(jié)點共享一個數(shù)據(jù)，可以采用廣播變量的方法。如果需要一些全局的聚合計算，可以使用累加器。

以上就是怎么給Spark傳遞函數(shù)，小編相信有部分知識點可能是我們?nèi)粘９ぷ鲿姷交蛴玫降?。希望你能通過這篇文章學(xué)到更多知識。更多詳情敬請關(guān)注億速云行業(yè)資訊頻道。