文章目錄
- 序章
- 第一個map. 以元素為粒度的資料轉換
- 點擊跳轉到下一講
序章
第一個map. 以元素為粒度的資料轉換
我們先來說說 map 算子的用法：給定映射函式 f，map(f) 以元素為粒度對 RDD 做資料轉換，其中 f 可以是帶有明確簽名的帶名函式，也可以是匿名函式，它的形參型別必須與 RDD 的元素型別保持一致，而輸出型別則任由開發者自行決定，
我們使用如下代碼，把包含單詞的 RDD 轉換成元素為（Key，Value）對的 RDD，后者統稱為 Paired RDD，
// 把普通RDD轉換為Paired RDD
val cleanWordRDD: RDD[String] = _
val kvRDD: RDD[(String, Int)] = cleanWordRDD.map(word => (word, 1))
在上面的代碼實作中，傳遞給 map 算子的形參，即：word => （word，1），就是我們上面說的映射函式 f，只不過，這里 f 是以匿名函式的方式進行定義的，其中左側的 word 表示匿名函式 f 的輸入形參，而右側的（word，1）則代表函式 f 的輸出結果，
如果我們把匿名函式變成帶名函式的話，可能你會看的更清楚一些，這里我用一段代碼重新定義了帶名函式 f，
// 把RDD元素轉換為（Key，Value）的形式
// 定義映射函式f
def f(word: String): (String, Int) = {
return (word, 1)
}
val cleanWordRDD: RDD[String] = _
val kvRDD: RDD[(String, Int)] = cleanWordRDD.map(f)
可以看到，我們使用 Scala 的 def 語法，明確定義了帶名映射函式 f，它的計算邏輯與剛剛的匿名函式是一致的，在做 RDD 資料轉換的時候，我們只需把函式 f 傳遞給 map 算子即可，不管 f 是匿名函式，還是帶名函式，map 算子的轉換邏輯都是一樣的，你不妨把以上兩種實作方式分別敲入到 spark-shell，去驗證執行結果的一致性，
到這里為止，我們就掌握了 map 算子的基本用法，現在你就可以定義任意復雜的映射函式 f，然后在 RDD 之上通過呼叫 map(f) 去翻著花樣地做各種各樣的資料轉換，
比如，通過定義如下的映射函式 f，我們就可以改寫 Word Count 的計數邏輯，也就是把“Spark”這個單詞的統計計數權重提高一倍：
// 把RDD元素轉換為（Key，Value）的形式
// 定義映射函式f
def f(word: String): (String, Int) = {
if (word.equals("Spark")) { return (word, 2) }
return (word, 1)
}
val cleanWordRDD: RDD[String] = _ // 請參考第一講獲取完整代碼
val kvRDD: RDD[(String, Int)] = cleanWordRDD.map(f)
盡管 map 算子足夠靈活，允許開發者自由定義轉換邏輯，不過，就像我們剛剛說的，map(f) 是以元素為粒度對 RDD 做資料轉換的，在某些計算場景下，這個特點會嚴重影響執行效率，為什么這么說呢？我們來看一個具體的例子，
比方說，我們把 Word Count 的計數需求，從原來的對單詞計數，改為對單詞的哈希值計數，在這種情況下，我們的代碼實作需要做哪些改動呢？我來示范一下：
// 把普通RDD轉換為Paired RDD
import java.security.MessageDigest
val cleanWordRDD: RDD[String] = _ // 請參考第一講獲取完整代碼
val kvRDD: RDD[(String, Int)] = cleanWordRDD.map{ word =>
// 獲取MD5物件實體
val md5 = MessageDigest.getInstance("MD5")
// 使用MD5計算哈希值
val hash = md5.digest(word.getBytes).mkString
// 回傳哈希值與數字1的Pair
(hash, 1)
}
由于 map(f) 是以元素為單元做轉換的，那么對于 RDD 中的每一條資料記錄，我們都需要實體化一個 MessageDigest 物件來計算這個元素的哈希值，
在工業級生產系統中，一個 RDD 動輒包含上百萬甚至是上億級別的資料記錄，如果處理每條記錄都需要事先創建 MessageDigest，那么實體化物件的開銷就會聚沙成塔，不知不覺地成為影響執行效率的罪魁禍首，
那么問題來了，有沒有什么辦法，能夠讓 Spark 在更粗的資料粒度上去處理資料呢？還真有，mapPartitions 和 mapPartitionsWithIndex 這對“孿生兄弟”就是用來解決類似的問題，相比 mapPartitions，mapPartitionsWithIndex 僅僅多出了一個資料磁區索引，因此接下來我們把重點放在 mapPartitions 上面，
點擊跳轉到下一講
轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/302745.html
標籤：其他
上一篇：淺談RabbitMQ