2. 程式人生 > >Spark函式詳解系列之RDD基本轉換+例項

Spark函式詳解系列之RDD基本轉換+例項

阿新 • • 發佈：2018-12-14

 RDD：彈性分散式資料集，是一種特殊集合 ‚ 支援多種來源 ‚ 有容錯機制 ‚ 可以被快取 ‚ 支援並行操作，一個RDD代表一個分割槽裡的資料集

  RDD有兩種操作運算元：

         Transformation（轉換）：Transformation屬於延遲計算，當一個RDD轉換成另一個RDD時並沒有立即進行轉換，僅僅是記住了資料集的邏輯操作

         Ation（執行）：觸發Spark作業的執行，真正觸發轉換運算元的計算

 

本系列主要講解Spark中常用的函式操作：

         1.RDD基本轉換

         2.鍵-值RDD轉換

         3.Action操作篇

本節所講函式

1.map(func)

2.flatMap(func)

3.mapPartitions(func)

4.mapPartitionsWithIndex(func)

5.simple(withReplacement,fraction,seed)

6.union(ortherDataset)

7.intersection(otherDataset)

8.distinct([numTasks])

9.cartesian(otherDataset)

10.coalesce(numPartitions，shuffle)

11.repartition(numPartition)

12.glom()

13.randomSplit(weight:Array[Double],seed)

 

基礎轉換操作：

 

1.map(func)：資料集中的每個元素經過使用者自定義的函式轉換形成一個新的RDD，新的RDD叫MappedRDD

（例1）

1

2

3

4

5

6

7

8

9

10

object Map {

  def main(args: Array[String]) {

    val conf = new SparkConf().setMaster("local").setAppName("map")

    val sc = new SparkContext(conf)

    val rdd = sc.parallelize(1 to 10)  //建立RDD

    val map = rdd.map(_*2)             //對RDD中的每個元素都乘於2

    map.foreach(x => print(x+" "))

    sc.stop()

  }

}

輸出：

2 4 6 8 10 12 14 16 18 20

(RDD依賴圖：紅色塊表示一個RDD區，黑色塊表示該分割槽集合，下同)

 

2.flatMap(func):與map類似，但每個元素輸入項都可以被對映到0個或多個的輸出項，最終將結果”扁平化“後輸出

（例2）

1

2

3

4

//...省略sc

   val rdd = sc.parallelize(1 to 5)

   val fm = rdd.flatMap(x => (1 to x)).collect()

   fm.foreach( x => print(x + " "))

輸出：

1 1 2 1 2 3 1 2 3 4 1 2 3 4 5

如果是map函式其輸出如下：

Range(1) Range(1, 2) Range(1, 2, 3) Range(1, 2, 3, 4) Range(1, 2, 3, 4, 5)

 (RDD依賴圖)

 

3.mapPartitions(func):類似與map，map作用於每個分割槽的每個元素，但mapPartitions作用於每個分割槽工

func的型別：Iterator[T] => Iterator[U]

假設有N個元素，有M個分割槽，那麼map的函式的將被呼叫N次,而mapPartitions被呼叫M次,當在對映的過程中不斷的建立物件時就可以使用mapPartitions比map的效率要高很多，比如當向資料庫寫入資料時，如果使用map就需要為每個元素建立connection物件，但使用mapPartitions的話就需要為每個分割槽建立connetcion物件

(例3)：輸出有女性的名字：

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

object MapPartitions {

//定義函式

  def partitionsFun(/*index : Int,*/iter : Iterator[(String,String)]) : Iterator[String] = {

    var woman = List[String]()

    while (iter.hasNext){

      val next = iter.next()

      next match {

        case (_,"female") => woman = /*"["+index+"]"+*/next._1 :: woman

        case _ =>

      }

    }

    return  woman.iterator

  }

 

  def main(args: Array[String]) {

    val conf = new SparkConf().setMaster("local").setAppName("mappartitions")

    val sc = new SparkContext(conf)

    val l = List(("kpop","female"),("zorro","male"),("mobin","male"),("lucy","female"))

    val rdd = sc.parallelize(l,2)

    val mp = rdd.mapPartitions(partitionsFun)

    /*val mp = rdd.mapPartitionsWithIndex(partitionsFun)*/

    mp.collect.foreach(x => (print(x +" ")))   //將分割槽中的元素轉換成Aarray再輸出

  }

}

輸出：

kpop lucy

其實這個效果可以用一條語句完成

1

val mp = rdd.mapPartitions(x => x.filter(_._2 == "female")).map(x => x._1)　

之所以不那麼做是為了演示函式的定義

  (RDD依賴圖)

 

4.mapPartitionsWithIndex(func):與mapPartitions類似，不同的時函式多了個分割槽索引的引數

func型別：(Int, Iterator[T]) => Iterator[U]

（例4）：將例3橙色的註釋部分去掉即是

輸出：（帶了分割槽索引）

[0]kpop [1]lucy

 

5.sample(withReplacement,fraction,seed):以指定的隨機種子隨機抽樣出數量為fraction的資料，withReplacement表示是抽出的資料是否放回，true為有放回的抽樣，false為無放回的抽樣

(例5)：從RDD中隨機且有放回的抽出50%的資料，隨機種子值為3（即可能以1 2 3的其中一個起始值）

1

2

3

4

5

//省略

    val rdd = sc.parallelize(1 to 10)

    val sample1 = rdd.sample(true,0.5,3)

    sample1.collect.foreach(x => print(x + " "))

    sc.stop

 

6.union(ortherDataset):將兩個RDD中的資料集進行合併，最終返回兩個RDD的並集，若RDD中存在相同的元素也不會去重

1

2

3

4

5

6

//省略sc

   val rdd1 = sc.parallelize(1 to 3)

   val rdd2 = sc.parallelize(3 to 5)

   val unionRDD = rdd1.union(rdd2)

   unionRDD.collect.foreach(x => print(x + " "))

   sc.stop　

輸出：

1 2 3 3 4 5

　　

7.intersection(otherDataset):返回兩個RDD的交集

1

2

3

4

5

6

//省略sc

val rdd1 = sc.parallelize(1 to 3)

val rdd2 = sc.parallelize(3 to 5)

val unionRDD = rdd1.intersection(rdd2)

unionRDD.collect.foreach(x => print(x + " "))

sc.stop　

輸出：

3 4

 

8.distinct([numTasks]):對RDD中的元素進行去重

1

2

3

4

5

//省略sc

val list = List(1,1,2,5,2,9,6,1)

val distinctRDD = sc.parallelize(list)

val unionRDD = distinctRDD.distinct()

unionRDD.collect.foreach(x => print(x + " "))　　

輸出：

1 6 9 5 2

 

9.cartesian(otherDataset):對兩個RDD中的所有元素進行笛卡爾積操作

1

2

3

4

5

//省略

val rdd1 = sc.parallelize(1 to 3)

val rdd2 = sc.parallelize(2 to 5)

val cartesianRDD = rdd1.cartesian(rdd2)

cartesianRDD.foreach(x => println(x + " "))　

輸出：

複製程式碼

(1,2)
(1,3)
(1,4)
(1,5)
(2,2)
(2,3)
(2,4)
(2,5)
(3,2)
(3,3)
(3,4)
(3,5)

複製程式碼

 (RDD依賴圖)

 

 

10.coalesce(numPartitions，shuffle):對RDD的分割槽進行重新分割槽，shuffle預設值為false,當shuffle=false時，不能增加分割槽數

目,但不會報錯，只是分割槽個數還是原來的

(例9:）shuffle=false

1

2

3

4

//省略 

val rdd = sc.parallelize(1 to 16,4)

val coalesceRDD = rdd.coalesce(3) //當suffle的值為false時，不能增加分割槽數(即分割槽數不能從5->7)

println("重新分割槽後的分割槽個數:"+coalesceRDD.partitions.size)　

輸出：

重新分割槽後的分割槽個數:3
//分割槽後的資料集
List(1, 2, 3, 4)
List(5, 6, 7, 8)
List(9, 10, 11, 12, 13, 14, 15, 16)

　

(例9.1:）shuffle=true

1

2

3

4

5

//...省略

val rdd = sc.parallelize(1 to 16,4)

val coalesceRDD = rdd.coalesce(7,true)

println("重新分割槽後的分割槽個數:"+coalesceRDD.partitions.size)

println("RDD依賴關係:"+coalesceRDD.toDebugString)　　

輸出：

複製程式碼

重新分割槽後的分割槽個數:5
RDD依賴關係:(5) MapPartitionsRDD[4] at coalesce at Coalesce.scala:14 []
| CoalescedRDD[3] at coalesce at Coalesce.scala:14 []
| ShuffledRDD[2] at coalesce at Coalesce.scala:14 []
+-(4) MapPartitionsRDD[1] at coalesce at Coalesce.scala:14 []
| ParallelCollectionRDD[0] at parallelize at Coalesce.scala:13 []
//分割槽後的資料集
List(10, 13)
List(1, 5, 11, 14)
List(2, 6, 12, 15)
List(3, 7, 16)
List(4, 8, 9)

複製程式碼

 (RDD依賴圖:coalesce(3,flase))

 

 

 (RDD依賴圖:coalesce(3,true))

 

 

11.repartition(numPartition):是函式coalesce(numPartition,true)的實現，效果和例9.1的coalesce(numPartition,true)的一樣

 

 

12.glom():將RDD的每個分割槽中的型別為T的元素轉換換陣列Array[T]

 

1

2

3

4

5

//省略

val rdd = sc.parallelize(1 to 16,4)

val glomRDD = rdd.glom() //RDD[Array[T]]

glomRDD.foreach(rdd => println(rdd.getClass.getSimpleName))

sc.stop　

輸出：

int[] //說明RDD中的元素被轉換成陣列Array[Int]

 

 

13.randomSplit(weight:Array[Double],seed):根據weight權重值將一個RDD劃分成多個RDD,權重越高劃分得到的元素較多的機率就越大

1

2

3

4

5

6

7

//省略sc

val rdd = sc.parallelize(1 to 10)

val randomSplitRDD = rdd.randomSplit(Array(1.0,2.0,7.0))

randomSplitRDD(0).foreach(x => print(x +" "))

randomSplitRDD(1).foreach(x => print(x +" "))

randomSplitRDD(2).foreach(x => print(x +" "))

sc.stop　

輸出：

2 4
3 8 9
1 5 6 7 10

 

 以上例子原始碼地址：https://github.com/Mobin-F/SparkExample/tree/master/src/main/scala/com/mobin/SparkRDDFun/TransFormation/KVRDD

相關推薦

Spark函式詳解系列之RDD基本轉換+例項

 RDD：彈性分散式資料集，是一種特殊集合 ‚ 支援多種來源 ‚ 有容錯機制 ‚ 可以被快取 ‚ 支援並行操作，一個RDD代表一個分割槽裡的資料集   RDD有兩種操作運算元：       &nbs

Spark函式詳解系列之RDD基本轉換

摘要：  RDD：彈性分散式資料集，是一種特殊集合 ‚ 支援多種來源 ‚ 有容錯機制 ‚ 可以被快取 ‚ 支援並行操作，一個RDD代表一個分割槽裡的資料集  RDD有兩種操作運算元：      Transformation（轉換）：Transformation屬於延遲計

Spark函數詳解系列之RDD基本轉換

9.png cal shuff reac 數組a water all conn data 摘要： RDD：彈性分布式數據集，是一種特殊集合 ? 支持多種來源 ? 有容錯機制 ? 可以被緩存 ? 支持並行操作，一個RDD代表一個分區裏的數據集 RDD有兩種操作算子： Tra

NTFS檔案系統詳解(一)之硬碟基本資訊

本文參考自部落格 一般硬碟正面貼有產品標籤，主要包括廠家資訊和產品資訊，如商標、型號、序列號、生產日期、容量、引數和主從設定方法等。這些資訊是正確使用硬碟的基本依據，下面將逐步介紹它們的含義。 硬碟主要由盤體、控制電路板和介面部件等組成，如圖1-1所示。

Java位元組碼詳解系列之二--解析位元組碼

1、javap檢視位元組碼內容上文介紹了位元組碼的結構，本文主要通過一個簡單的例子說明class位元組碼的每一個欄位。package com.zcm.test; import java.io.Serializable; public class SourceTest im

ThinkPHP函式詳解系列--單字母函式

ThinkPHP中有很多重要的函式，其中大部分是單字母函式，這些函式在開發體驗中發揮了重大的作用。 為了能方便大家學習和掌握，在這裡彙總下ThinkPHP中的經典函式用法，無論你是資深還是菜鳥TPer，絕對不容錯過哦 ^_^ A 函式：例項化控制器 R 函式：直接呼叫控

C# 9.0新特性詳解系列之三：模組初始化器

## [1][1] 背景動機 關於模組或者程式集初始化工作一直是C#的一個痛點，微軟內部外部都有大量的報告反應很多客戶一直被這個問題困擾，這還不算沒有統計上的客戶。那麼解決這個問題，還有基於什麼樣的考慮呢？ * 在庫載入的時候，能以最小的開銷、無需使用者顯式呼叫任何介面，使客戶做一些期望的和一次性的初始化。

C#9.0新特性詳解系列之四：頂級程式語句（Top-Level Programs)

## 1 背景與動機 通常，如果只想用C#在控制檯上列印一行“Hello World！”，這可不是Console.WriteLine("Hello World!");一條語句就可以搞定的，還涉及到其他必要基礎程式碼（如定義類和入口函式Main），例如下面： ```C# using System; clas

C# 9.0新特性詳解系列之五：記錄(record)和with表示式

## [1][1] 背景與動機 傳統面向物件程式設計的核心思想是一個物件有著唯一標識，表現為物件引用，封裝著隨時可變的屬性狀態，如果你改變了一個屬性的狀態，這個物件還是原來那個物件，就是物件引用沒有因為狀態的改變而改變，也就是說該物件可以有很多種狀態。C#從最初開始也是一直這樣設計和工作的。但是一些時候，你

C#9.0新特性詳解系列之六：增強的模式匹配

自C#7.0以來，模式匹配就作為C#的一項重要的新特性在不斷地演化，這個借鑑於其小弟F#的函數語言程式設計的概念，使得C#的本領越來越多，C#9.0就對模式匹配這一功能做了進一步的增強。 為了更為深入和全面的瞭解模式匹配，在介紹C#9.0對模式匹配增強部分之前，我對模式匹配整體做一個回顧。 ## 1

python畫圖之散點圖sactter函式詳解，基本雜湊點繪製

原文地址：http://blog.csdn.net/u013634684/article/details/49646311 http://blog.csdn.net/abcjennifer/article/details/19848269 http://blog.csdn.

Spark RDD操作：combineByKey函式詳解

當資料集一鍵值對形式組織的時候，聚合具有相同鍵的元素進行一些統計是很常見的操作。對於Pair RDD常見的聚合操作如：reduceByKey，foldByKey，groupByKey，combineByKey。這裡重點要說的是combineByKey。因為combineBy

Spark核心RDD：foldByKey函式詳解

foldByKey(zeroValue: V, numPartitions: Int)(func: (V, V) => V): RDD[(K, V)]foldByKey(zeroValue: V

JDBC詳解系列（二）之加載驅動

red mar mys ons try path 替換 host man ---[來自我的CSDN博客](http://blog.csdn.net/weixin_37139197/article/details/78838091)--- ??在JDBC詳解系列（一）之流程中

爬蟲之刃----趕集網招聘類爬取案例詳解(系列四)

前言 本篇承襲之前的系列文章，開始動真格。以趕集網招聘類資訊爬取為例，詳細解說爬蟲程式構建過程。 準備工作： 閱讀之前的系列一、系列二、系列三，有一定遞進關係 登陸趕集網，瞭解下“地形” OK，let’s go！ 構建URL庫

JS之函式詳解

1、定義 函式也是一個物件；它是可以重複執行的程式碼塊；是可以完成特定功能的一段程式碼；使用typeof檢查一個函式物件時，會返回function。 2、作用 因為一部分程式碼使用次數可能會很多，所以封裝起來，需要的時候呼叫就可以了。這樣做可以將複雜的事情簡單化 ，將重複使用的程式碼進行

Matlab之graythresh()函式詳解

1 函式簡介 函式功能：使用 最大類間方差法找到圖片的一個合適的閾值（threshold）。在使用 im2bw函式將灰度影象轉換為二值影象時，需要設定一個閾值，這個函式可以幫助我們獲得一個合適的閾值。利用這個閾值通常比人為設定的閾值能更好地把一張灰度影象轉換為二值影象

Linux I/O複用之select函式詳解

置頂 2017年02月12日 20:50:08 難免有錯_ 閱讀數：7438更多 select函式的功能和呼叫順序 使用select函式時統一監視多個檔案描述符的： 1、 是否存在套接字接收資料？ 2、 無需阻塞傳輸資料的套接字有哪些? 3、 哪些套接字發生了

docker系列詳解<一>之docker安裝

1、Docker 要求 CentOS 系統的核心版本高於 3.10 ，檢視本頁面的前提條件來驗證你的CentOS 版本是否支援 Docker 。 通過 uname -r 命令檢視你當前的核心版本 $ uname -r 2、使用 root 許可權登入 Centos。確保 yum 包更新到最新。 $ su

Android NDK——必知必會之JNI的C++操作函式詳解和小結（三）

引言 上一篇講解了一些關於JNI和NDK的必知必會的理論知識和機制，由於篇幅問題把關於JNI的重要的函式放到這篇，具體使用留到下一篇，此係列文章基連結： 一、JNI中的函式概述 在JNI層我們基本上都是通過env指標來呼叫jni.h標頭檔案裡定義的函式，JNI