Spark RDD Scala語言程式設計

RDD（Resilient Distributed Dataset）是一個不可變的分散式物件集合， 每個rdd被分為多個分割槽， 這些分割槽執行在叢集的不同節點上。rdd支援兩種型別的操作：轉化(trainsformation)和行動(action)， Spark只會惰性計算rdd, 也就是說， 轉化操作的rdd不會立即計算， 而是在其第一次遇到行動操作時才去計算， 如果想在多個行動操作中複用一個rdd, 可以使用RDD.persist()讓Spark把這個rdd快取下來。

0. 初始化SparkContext

import org.apache.spark.{SparkConf, SparkContext}

val
 
 sc = new SparkContext(new SparkConf().setMaster("local[*]").setAppName("spark-rdd-demo"))

1. 建立RDD

Spark提供了2種建立RDD的方式：

1.1 讀取外部資料集

val lines = sc.parallelize(List("Java", "Scala", "C++"))

1.2 在驅動器程式中對一個集合進行並行化

val lines = sc.textFile("hdfs://dash-dev:9000/input/test.txt")

2. RDD操作

2.1 轉化操作

RDD的轉化操作是返回新RDD的操作, 常用轉化操作總結如下：

表1: 對一個數據為{1,2,3,3}的RDD進行基本的轉化操作

表2: 對資料分別為{1,2,3}和{2,3,4}RDD進行鍼對2個RDD的轉化操作

2.2 行動操作

RDD的行動操作會把最終求得的結果返回驅動器程式， 或者寫入外部儲存系統中。

表3: 對一個數據為{1,2,3,3}的RDD進行基本RDD的行動操作

2.3 向Spark傳遞函式

Spark的大部分轉化和行動操作都要依賴於使用者傳遞的函式來計算， 當傳遞的物件是某個物件的成員， 或者包含了對某個物件中一個欄位的引用時(如self.field), Spark就會把整個物件傳送到工作節點上－－這比你本意想傳遞的東西大太多了！替代的方案是，把你需要的欄位從物件中拿出來放到一個區域性變數中， 然後傳遞這個區域性變數：

class SearchFunctions(val query: String) {
    def isMatch(s: String): Boolean = {
        s.contains(query)
    }

    def getMatchesFunctionReference(rdd: RDD[String]): RDD[String] = {
        // 問題："isMatch"表示"this.isMatch", 因此會傳遞整個this
        rdd.map(isMatch)
    }

    def getMatchesFieldReference(rdd: RDD[String]): RDD[String] = {
        // 問題： "query"表示"this.query", 因此會傳遞整個this
        rdd.map(x => x.split(query))
    }

    def getMatchesNoReference(rdd: RDD[String]): RDD[String] = {
        // 安全：只把我們需要的欄位拿出來放入區域性變數中
        val localQuery = this.query
        rdd.map(x => x.split(localQuery))
    }
}

另外， 要注意的是， Spark要求我們傳入的函式及其應用的資料是可序列化的(實現了Java的Serializable介面)， 否則會出現NotSerializableException。