RDD的轉換操作,分三種:單value,雙value交互,(k,v)對

来源:https://www.cnblogs.com/liangyan131/archive/2019/12/10/12019351.html

import org.apache.spark.rdd.RDDimport org.apache.spark.{Partitioner, SparkConf, SparkContext} object Transformation { def main(args: Array[String]): U ...


import org.apache.spark.rdd.RDD
import org.apache.spark.{Partitioner, SparkConf, SparkContext}
object Transformation {

  def main(args: Array[String]): Unit = {

    val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("Transformation")

    val sc = new SparkContext(config)

    val listRDD = sc.makeRDD(1 to 10)
    val listRDD2 = sc.makeRDD(Array(List(1, 2), List(3, 4)))
    val listRDD3 = sc.makeRDD(5 to 14)

    /***************************單value*****************************/

    /**
      * map(func)
      * 每次處理1條數據
      */

//    val mapRDD = listRDD.map(_ * 2)

    /**
      * mapPartitions(func)
      * 每次處理一組分區數據,效率高,但可能出現記憶體溢出(因為處理完一組分區後再釋放)
      */

//     val mapPartitionsRDD = listRDD.mapPartitions(datas=>{
//       datas.map(data => data * 2)
//     })


    /**
      * mapPartitionsWithIndex(func)
      * 函數的輸入多了分區號
      */

//    val tupleRDD: RDD[(Int, String)] = listRDD.mapPartitionsWithIndex {
//      case (num, datas) => {
//        datas.map((_, " 分區號:" + num))
//      }
//    }

    /**
      *  flatMap(func)
      *  將map後的數據扁平
      */

//    val flatMAPRDD: RDD[Int] = listRDD2.flatMap(datas => datas)

    /**
      * glom()
      * 將一個分區的數據放在一個數組裡
      */

//    val glomRDD: RDD[Array[Int]] = listRDD.glom()

    /**
      * groupBy(func)
      * 按照函數的返回值進行分組,分組後的數據(K:分組的key,V:分組的集合)
      */

//    val groupByRDD: RDD[(Int, Iterable[Int])] = listRDD.groupBy(i => i%2)
//    groupByRDD.collect().foreach(println)

    /**
      * filter(func)
      * 按照返回值為true的過濾
      */

//    val filterRDD: RDD[Int] = listRDD.filter(x => x % 2 ==0)
//    filterRDD.collect().foreach(println)

    /**
      * sample(withReplacement : scala.Boolean, fraction : scala.Double, seed : scala.Long)
      * 隨機抽樣
      */

//    val sampleRDD: RDD[Int] = listRDD.sample(false, 0.4, 1)
//    sampleRDD.collect().foreach(println)

    /**
      * distinct()
      * 去重,且去重後會shuffler,可以指定去重後的分區數
      */

//    val distinctRDD: RDD[Int] = listRDD.distinct()
//    distinctRDD.collect().foreach(println)

    /**
      * coalesce(n)
      * 縮減分區的數量,可以簡單的理解為合併分區,預設,沒有shuffler,可以加參數true指定shuffler
      */

//    println("縮減分區前 = " + listRDD.partitions.size)
//    val coalesceRDD: RDD[Int] = listRDD.coalesce(2)
//    println("縮減分區前 = " + coalesceRDD.partitions.size)

    /**
      * repartition()
      * 重新分區,有shuffler。它其實就是帶true的coalesce
      */

//    listRDD.glom().collect().foreach(arrays => {
//      println(arrays.mkString(","))
//    })
//    val repartitionRDD: RDD[Int] = listRDD.repartition(2)
//    repartitionRDD.glom().collect().foreach(arrays => {
//      println(arrays.mkString(","))
//    })

    /**
      * sortBy(f: (T) => K,ascending: Boolean = true,numPartitions: Int = this.partitions.length))
      * 根據函數排序
      */

//    val sortByRDD: RDD[Int] = listRDD.sortBy(n => n % 2, false)
//    sortByRDD.collect().foreach(println)

    /**************************雙value交互*****************************/

    /**
      * 雙value交互
      * A.union(B)         對A、B合併。(不去重)
      * A.subtract(B)      對A減去和B中的相同的
      * A.cartesian(B)     對A、B求笛卡爾乘積
      * A.zip(B)           將A、B組成(k,v),個數、分區數要相等
      * A.union(B) 對A、B求並集
      */

//    listRDD.union(listRDD3).collect().foreach(println)
//    listRDD.subtract(listRDD3).collect().foreach(println)
//    listRDD.intersection(listRDD3).collect().foreach(println)
//    listRDD.cartesian(listRDD3).collect().foreach(println)
//    listRDD.zip(listRDD3).collect().foreach(println)


    /**************************(k,v)對*******************************/

    val pairRDD1: RDD[(Int, String)] = sc.parallelize(Array((1, "aa"), (1, "bb"), (3, "cc"), (3, "dd")),  4)
    val pairRDD2: RDD[(String, Int)] = sc.parallelize(Array(("a", 3), ("a", 2), ("c", 4),
                                                            ("b", 3), ("c", 6), ("c", 8)),  2)
    val pairRDD3: RDD[(Int, String)] = sc.parallelize(Array((1, "zzz"), (3, "xxx")))

    /**
      * partitionBy(partitioner: Partitioner)
      * 按照分區器進行分區
      */

//    pairRDD1.partitionBy(new org.apache.spark.HashPartitioner(2))
//      .glom().collect().foreach(arrays => {
//      println(arrays.mkString(","))
//    })

//    pairRDD1.partitionBy(new MyPartitioner(3))
//      .glom().collect().foreach(arrays => {
//      println(arrays.mkString(","))
//    })

    /**
      * groupByKey()
      * 單純把key相等的value放在一起,生成序列
      */
//    pairRDD1.groupByKey().collect().foreach(println)


    /**
      * reduceByKey(func)
      * 按key聚合,並且按函數對key相等的value進行操作
      */

//    pairRDD1.reduceByKey(_ + _)
//      .glom().collect().foreach(arrays => {
//      println(arrays.mkString(","))
//    })


    /**
      * aggregateByKey[U: ClassTag](zeroValue: U)(seqOp: (U, V) => U, combOp: (U, U) => U)
      * zeroValue:每個分區的每一個key的初始值
      * seqOp:每個分區里的聚合函數
      * seqOp:分區間的聚合函數
      */


    // 取出每個分區相同對key的最大值,在相加
//    pairRDD2.aggregateByKey(0)(math.max(_,_), _+_)
//      .glom().collect().foreach(arrays => {
//      println(arrays.mkString(","))
//    })

    /**
      * foldByKey(zeroValue: V)(func: (V, V) => V)
      * 其實就是aggregateByKey的簡化版,seqOp和seqOp相同
      */

//    pairRDD2.foldByKey(0)(_ + _)
//      .glom().collect().foreach(arrays => {
//      println(arrays.mkString(","))
//    })

    /**
      * combineByKey[C](
      * createCombiner: V => C,
      * mergeValue: (C, V) => C,
      * mergeCombiners: (C, C) => C,
      * partitioner: Partitioner,
      * mapSideCombine: Boolean = true,
      * serializer: Serializer = null)
      *
      * 主要就是比aggregateByKey多了一個createCombiner,用於計算初始值
      */

    // 計算相同key的value的均值
//    pairRDD2.combineByKey(
//      (_, 1),
//      (acc:(Int, Int), v) => (acc._1 + v, acc._2 + 1),
//      (acc1:(Int, Int), acc2:(Int, Int)) => (acc1._1 + acc2._1, acc1._2 + acc2._2))
//      .map{case (key, value) => (key, value._1 / value._2.toDouble)}
//      .collect().foreach(println)

    /**
      * sortByKey(ascending: Boolean = true, numPartitions: Int = self.partitions.length)
      * 按key排序
      */

//    pairRDD1.sortByKey(true)
//      .collect().foreach(println)


    /**
      * mapValues(func)
      * 只對value做轉換
      */

//    pairRDD1.mapValues(value => value + "|||")
//      .collect().foreach(println)

    /**
      * A.join(B, numP)
      * 把key相同的value組合在一起(性能較低)
      */

//    pairRDD1.join(pairRDD3)
//      .collect().foreach(println)

    /**
      * A.cogroup(B)
      * (k, v1) 和 (k, v2)cogroup 後,得到(k, v1集合,v2集合)
      */

    pairRDD1.cogroup(pairRDD3)
      .collect().foreach(println)

    sc.stop()

  }
}

// 自定義分區器
class MyPartitioner (partitions: Int) extends Partitioner {
  override def numPartitions: Int = {
    partitions
  }

  override def getPartition(key: Any): Int = {
    1
  }
}

  //只寫代碼不讓我發出來--忽略這一行


您的分享是我們最大的動力!

更多相關文章
  • '''''' ''' 1、簡述面向對象三大特性並用示例解釋說明?【背寫】 1、封裝 狹義的封裝:把一組屬性封裝到一個對象,創建對象的時候 廣義的封裝:代碼塊,函數、對象、類、模塊-py文件都是封裝 把封裝後的對象看成一個黑盒子,只需要關註輸入和輸出,不必關註黑盒子內部的實現 2、繼承 1、避免代..... ...
  • 今天在使用go與php的AES加解密交互中,一直有個問題那就是在go中加密後,在php端始終都是無法解密,經過排查最後發現是加密key長度引起的問題, 這裡簡單記錄下。 go的AES使用的是第三方的庫, "openssl" ,因為用的匆忙,沒註意看文檔,所以就直接弄了示例代碼,才發現和php端無法解 ...
  • 此模式通過一個模板方法來定義程式的框架或演算法,通常模板方法定義在基類中,即原始的模板,然後子類就可以根據不同的需要實現或重寫模板方法中的某些演算法步驟或者框架的某部分,最後達到使用相同模板實現不同功能的效果。 核心思想: 使用一個模板方法定義好總的演算法框架。 子類中根據需要重新定義某些操作,但是不能修 ...
  • 有個需求,從某個介面下載的一個zip壓縮包,往裡面添加一個說明文件。搜索了一下,沒有找到往zip直接添加文件的方法,最終解決方法是先解壓、再壓縮。具體過程如下: ...
  • 屬性 語法格式:修飾符 類型 屬性名 = 初值; 說明: 修飾符:public、protected、private:用於表示成員變數的訪問許可權。static:表示該成員變數為類變數,也稱為靜態變數。final:表示將該成員變數聲明為常量,其值無法更改。 類型:表示變數的類型。 屬性名:表示變數名稱。 ...
  • Java類的初始化順序 多說無益,以下是本人親自試驗的代碼,一目瞭然: 1 package test1; 2 3 public class Test { 4 public static void main(String[] argc) { 5 new Child(); 6 System.out.pr ...
  • composer install thinkphp6 報錯 Parse error: syntax error, unexpected ':', expecting '{' in vendor\topthink\think-helper\src\helper.php on line 233 ...
  • 前言本文的文字及圖片來源於網路,僅供學習、交流使用,不具有任何商業用途,版權歸原作者所有,如有問題請及時聯繫我們以作處理。作者:萬能搜吧 都是copy的百度SDK文檔,簡單說說怎麼用。 1、沒安裝Python的參見此文:Python學習筆記系列 1 ——安裝調試Python開發軟體 2、win+r輸 ...
一周排行
  • 比如要拆分“呵呵呵90909086676喝喝999”,下麵當type=0返回的是中文字元串“呵呵呵,喝喝”,type=1返回的是數字字元串“90909086676,999”, private string GetStrings(string str,int type=0) { IList<strin ...
  • Swagger一個優秀的Api介面文檔生成工具。Swagger可以可以動態生成Api介面文檔,有效的降低前後端人員關於Api介面的溝通成本,促進項目高效開發。 1、使用NuGet安裝最新的包:Swashbuckle.AspNetCore。 2、編輯項目文件(NetCoreTemplate.Web.c ...
  • 2020 年 7 月 30 日, 由.NET基金會和微軟 將舉辦一個線上和為期一天的活動,包括 微軟 .NET 團隊的演講者以及社區的演講者。本次線上大會 專註.NET框架構建微服務,演講者分享構建和部署雲原生應用程式的最佳實踐、模式、提示和技巧。有關更多信息和隨時瞭解情況:https://focu... ...
  • #abp框架Excel導出——基於vue #1.技術棧 ##1.1 前端採用vue,官方提供 UI套件用的是iview ##1.2 後臺是abp——aspnetboilerplate 即abp v1,https://github.com/aspnetboilerplate/aspnetboilerp ...
  • 前言 本文的文字及圖片來源於網路,僅供學習、交流使用,不具有任何商業用途,版權歸原作者所有,如有問題請及時聯繫我們以作處理。 作者:碧茂大數據 PS:如有需要Python學習資料的小伙伴可以加下方的群去找免費管理員領取 input()輸入 Python提供了 input() 內置函數從標準輸入讀入一 ...
  • 從12年到20年,python以肉眼可見的趨勢超過了java,成為了當今It界人人皆知的編程語言。 python為什麼這麼火? 網路編程語言搜索指數 適合初學者 Python具有語法簡單、語句清晰的特點,這就讓初學者在學習階段可以把精力集中在編程對象和思維方法上。 大佬都在用 Google,YouT ...
  • 在社會上存在一種普遍的對培訓機構的學生一種歧視的現象,具體表現在,比如:當你去公司面試的時候,一旦你說了你是培訓機構出來的,那麼基本上你就涼了,那麼你瞞著不說,然後又通過了面試成功入職,但是以後一旦在公司被髮現有培訓經歷,可能會面臨被降薪,甚至被辭退,培訓機構出來的學生,在用人單位眼裡就是能力低下的 ...
  • from typing import List# 這道題看了大佬寫的代碼,經過自己的理解寫出來了。# 從最外圍的四周找有沒有為O的,如果有的話就進入深搜函數,然後深搜遍歷# 判斷上下左右的位置是否為Oclass Solution: def solve(self, board: List[List[s ...
  • import requests; import re; import os; # 1.請求網頁 header = { "user-agent":'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_5) AppleWebKit/537.36 (KHTML, li ...
  • import requests; import re; import os; import parsel; 1.請求網頁 header = { "user-agent":'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_5) AppleWebKit/537. ...