2. 程式人生 > >跟我一起學Spark之——RDD Join中寬依賴與窄依賴的判斷

跟我一起學Spark之——RDD Join中寬依賴與窄依賴的判斷

阿新 • • 發佈：2018-11-11

1.規律

　　　如果JoinAPI之前被呼叫的RDD API是寬依賴(存在shuffle), 而且兩個join的RDD的分割槽數量一致，join結果的rdd分割槽數量也一樣，這個時候join api是窄依賴
　　除此之外的，rdd 的join api是寬依賴

2.Join的理解

　

3.舉例

A表資料：
  1 a
  2 b
  3 c
B表資料：
  1 aa1
  1 aa2
  2 bb1
  2 bb2
  2 bb3
  4 dd1

A inner join B:
  1    a 1 aa1
  1    a 1 aa2
  2　　 b 2 bb1
  2 　　b 2 bb2
  2 　　b 2 bb3

A left outer join B:
  1    a 1 aa1
  1    a 1 aa2
  2　　 b 2 bb1
  2　　 b 2 bb2
  2　　 b 2 bb3
  3　　 c null null

A right outer join B:
  1    a 1 aa1
  1    a 1 aa2
  2 　　b 2 bb1
  2 　　b 2 bb2
  2 　　b 2 bb3
  null null 4 dd1

A full outer join B:
  1    a 1 aa1
  1    a 1 aa2
  2 　　b 2 bb1
  2 　　b 2 bb2
  2 　　b 2 bb3
  3 　　c null null
  null null 4 dd1

A left semi join B:
  1 a
  2 b

4.API

　　必須是Key/value鍵值對

　　

5.測試程式

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

/**
  * RDD資料Join相關API講解
  * Created by ibf on 02/09.
  */
object RDDJoin {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
      .setMaster("local[*]")
      .setAppName("RDD-Join")
    val sc = SparkContext.getOrCreate(conf)

    // ==================具體程式碼======================
    // 模擬資料產生
    val rdd1 = sc.parallelize(Array(
      (1, "張三1"),
      (1, "張三2"),
      (2, "李四"),
      (3, "王五"),
      (4, "Tom"),
      (5, "Gerry"),
      (6, "莉莉")
    ), 1)

    val rdd2 = sc.parallelize(Array(
      (1, "上海"),
      (2, "北京1"),
      (2, "北京2"),
      (3, "南京"),
      (4, "紐約"),
      (6, "深圳"),
      (7, "香港")
    ), 1)

    // 呼叫RDD API實現內連線
    val joinResultRDD = rdd1.join(rdd2).map {
      case (id, (name, address)) => {
        (id, name, address)
      }
    }
    println("----------------")
    joinResultRDD.foreachPartition(iter => {
      iter.foreach(println)
    })
    // 呼叫RDD API實現左外連線
    val leftJoinResultRDd = rdd1.leftOuterJoin(rdd2).map {
      case (id, (name, addressOption)) => {
        (id, name, addressOption.getOrElse("NULL"))
      }
    }
    println("----------------")
    leftJoinResultRDd.foreachPartition(iter => {
      iter.foreach(println)
    })
    // 左外連線稍微變化一下：需要左表出現，右表不出現的資料(not in)
    println("----------------")
    rdd1.leftOuterJoin(rdd2).filter(_._2._2.isEmpty).map {
      case (id, (name, _)) => (id, name)
    }.foreachPartition(iter => {
      iter.foreach(println)
    })

    // 右外連線
    println("----------------")
    rdd1
      .rightOuterJoin(rdd2)
      .map {
        case (id, (nameOption, address)) => {
          (id, nameOption.getOrElse("NULL"), address)
        }
      }
      .foreachPartition(iter => iter.foreach(println))

    // 全外連線
    println("----------------")
    rdd1
      .fullOuterJoin(rdd2)
      .map {
        case (id, (nameOption, addressOption)) => {
          (id, nameOption.getOrElse("NULL"), addressOption.getOrElse("NULL"))
        }
      }
      .foreachPartition(iter => iter.foreach(println))

    // 休眠為了看4040頁面
        Thread.sleep(1000000)
  }
}

6.說明　

　RDD join API:
　　def join[W](other: RDD[(K, W)]): RDD[(K, (V, W))]
　　　　返回值是RDD，RDD中的型別是一個二元組(a)，a第一個元素是KEY型別的值(join的key), a第二個元素又是二元組(b), b的第一個元素是來自呼叫join函式的RDD的value，
　　　　b的第二個元素是來自引數other這個RDD的value

　　def leftOuterJoin[W](other: RDD[(K, W)]): RDD[(K, (V, Option[W]))]
　　　　對於右邊的資料返回的是Option型別是資料，所以如果右表資料不存在，返回的是None；否則是一個Some的具體資料

　　def rightOuterJoin[W](other: RDD[(K, W)]): RDD[(K, (Option[V], W))]
　　　　對於左邊的資料返回的是Option型別是資料，所以如果左表資料不存在，返回的是None；否則是一個Some的具體資料

　　def fullOuterJoin[W](other: RDD[(K, W)]): RDD[(K, (Option[V], Option[W]))]
　　　　返回的value型別是Option封裝後的資料，如果資料不存在, 返回的是None，存在返回的是Some具體資料

7.缺點

8.優化程式

　　沒有使用API，根據原理寫一個。

　　減少shufflw運算元的使用。

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

/**
  * RDD資料Join相關API講解
  * Created by ibf on 02/09.
  */
object RDDJoin {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
      .setMaster("local[*]")
      .setAppName("RDD-Join")
    val sc = SparkContext.getOrCreate(conf)

    // ==================具體程式碼======================
    // 模擬資料產生
    val rdd1 = sc.parallelize(Array(
      (1, "張三1"),
      (1, "張三2"),
      (2, "李四"),
      (3, "王五"),
      (4, "Tom"),
      (5, "Gerry"),
      (6, "莉莉")
    ), 1)

    val rdd2 = sc.parallelize(Array(
      (1, "上海"),
      (2, "北京1"),
      (2, "北京2"),
      (3, "南京"),
      (4, "紐約"),
      (6, "深圳"),
      (7, "香港")
    ), 1)

    // 假設rdd2的資料比較少，將rdd2的資料廣播出去
    val leastRDDCollection = rdd2.collect()
    val broadcastRDDCollection = sc.broadcast(leastRDDCollection)

    println("++++++++++++++++++")
    // 類似Inner Join的操作，Inner Join的功能：將兩個表都出現的資料合併
    println("-------------------")
    rdd1
      // 過濾rdd1中的資料，只要在rdd1中出現的資料，沒有出現的資料過濾掉
      .filter(tuple => broadcastRDDCollection.value.map(_._1).contains(tuple._1))
      // 資料合併，由於一條rdd1的資料可能在rdd2中存在多條對應資料，所以使用flatMap
      .flatMap {
      case (id, name) => {
        broadcastRDDCollection.value.filter(_._1 == id).map {
          case (_, address) => {
            (id, name, address)
          }
        }
      }
    }
      .foreachPartition(iter => iter.foreach(println))

    // 左外連線
    println("---------------------")
    rdd1
      .flatMap {
        case (id, name) => {
          // 從右表所屬的廣播變數中獲取對應id的集合列表
          val list = broadcastRDDCollection.value.filter(_._1 == id)
          // 對應id的集合可能為空，也可能資料有多個
          if (list.nonEmpty) {
            // 存在多個
            list.map(tuple => (id, name, tuple._2))
          } else {
            // id在右表中不存在，填預設值
            (id, name, "NULL") :: Nil
          }
        }
      }
      .foreachPartition(iter => iter.foreach(println))

    // 右外連線
    /**
      * rdd2中所有資料出現，由於rdd2中的資料在driver中可以儲存，可以認為rdd1和rdd2通過right join之後的資料也可以在driver中儲存下
      **/
    println("---------------------")
    // 將rdd1中符合條件的資料過濾出來儲存到driver中
    val stage1 = rdd1
      .filter(tuple => broadcastRDDCollection.value.map(_._1).contains(tuple._1))
      .collect()
    // 將driver中兩個集合進行right join
    val stage2 = leastRDDCollection.flatMap {
      case (id, address) => {
        val list = stage1.filter(_._1 == id)
        if (list.nonEmpty) {
          list.map(tuple => (id, tuple._2, address))
        } else {
          Iterator.single((id, "NULL", address))
        }
      }
    }
    stage2.foreach(println)

    // TODO: 全外連線，不寫程式碼，因為程式碼比較複雜

    //====================================
    // 左半連線：只出現左表資料(要求資料必須在右表中也出現過)，如果左表的資料在右表中出現多次，最終結果只出現一次
    println("+++++++++++++++++")
    println("-----------------------")
    rdd1
      .join(rdd2)
      .map {
        case (id, (name, _)) => (id, name)
      }
      .distinct()
      .foreachPartition(iter => iter.foreach(println))
    println("------------------------")
    rdd1
      .filter(tuple => broadcastRDDCollection.value.map(_._1).contains(tuple._1))
      .foreachPartition(iter => iter.foreach(println))

    // 休眠為了看4040頁面
        Thread.sleep(1000000)
  }
}

9.Join的窄依賴程式

　　使用reduceByKey，裡面的程式會給一個分割槽。 

package com.ibeifeng.senior.join

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

/**
  * RDD資料Join相關API講解
  * Created by ibf on 02/09.
  */
object RDDJoin2 {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
      .setMaster("local[*]")
      .setAppName("RDD-Join")
    val sc = SparkContext.getOrCreate(conf)

    // ==================具體程式碼======================
    // 模擬資料產生， 新增map、reduceByKey、mapPartitions等api的主要功能是給rdd1和rdd2中新增一個分割槽器(表示當前rdd是存在shuffle過程的)
    val rdd1 = sc.parallelize(Array(
      (1, "張三1"),
      (1, "張三2"),
      (2, "李四"),
      (3, "王五"),
      (4, "Tom"),
      (5, "Gerry"),
      (6, "莉莉")
    ), 1).map(x => (x, null)).reduceByKey((x,y) => x, 1).mapPartitions(
      iter => iter.map(tuple => tuple._1),
      true // 使用上一個RDD的分割槽器，false表示不使用, 設定為None
    )

    val rdd2 = sc.parallelize(Array(
      (1, "上海"),
      (2, "北京1"),
      (2, "北京2"),
      (3, "南京"),
      (4, "紐約"),
      (6, "深圳"),
      (7, "香港")
    ), 1).map(x => (x, null)).reduceByKey((x,y) => x, 1).mapPartitions(
      iter => iter.map(tuple => tuple._1),
      true // 使用上一個RDD的分割槽器，false表示不使用, 設定為None
    )

    // 呼叫RDD API實現內連線
    val joinResultRDD = rdd1.join(rdd2).map {
      case (id, (name, address)) => {
        (id, name, address)
      }
    }
    println("----------------")
    joinResultRDD.foreachPartition(iter => {
      iter.foreach(println)
    })

    // 休眠為了看4040頁面
        Thread.sleep(1000000)
  }
}

 

原文連結：https://www.cnblogs.com/juncaoit/p/6528146.html

相關推薦

跟我一起學Spark之——RDD Join中寬依賴與窄依賴的判斷

1.規律 　　　如果JoinAPI之前被呼叫的RDD API是寬依賴(存在shuffle), 而且兩個join的RDD的分割槽數量一致，join結果的rdd分割槽數量也一樣，這個時候join api是窄依賴 　　除此之外的，rdd 的join api是寬依賴 2.Join的理解 　

跟我一起學Spark之——《Spark快速大資料分析》pdf版下載

連結：https://pan.baidu.com/s/1vjQCJLyiXzIj6gnCCDyv3g  提取碼：ib01 國慶第四天，去逛了半天的王府井書店，五層出電梯右邊最裡面，倒數第三排《資料結構》，找到了一本很不錯的書《Spark快速大資料分析》，試讀了下，我很喜歡，也很適合

跟我一起學Spark之——Windows10下spark2.3.0本地開發環境搭建-親測

相關元件版本： JDK1.8.0_171，hadoop-2.7.6，Spark-2.3.0，Scala-2.11.8，Maven-3.5.3，ideaIC-2018.1.4.exe，spark-2.3.0-bin-hadoop2.7 1.1  JDK1.8.0_171 a.&n

跟我一起學Spark之——資料分割槽

前言         控制資料分佈以獲得最少的網路傳輸可以極大地提升整體效能。         如果給定RDD只需要被掃描一次（例如大小表join中的小表），我們完全沒有必要對其預先進行分割槽處理，只有當資料

跟我一起學MongoDB之——視覺化工具Compass的簡單使用

首先官方為大家介紹了一下MongoDB Compass的主要功能： 1.對資料的視覺化 2.插入、修改、刪除 3.除錯、優化 進入Compass首先出現連線介面如下（此處我使用的是我操作時候的介面狀態） 直接使用預設None無賬戶連線了嘻嘻嘻 點選綠色CONNEC

跟我一起學C++之從C到C++（bool型別）

bool型別 C++新增型別，表示邏輯真與假 1.邏輯型也稱布林型，其取值為true（邏輯真）和false（邏輯假），儲存位元組數在不同編譯系統中可能有所不同，VC++中為1個位元組。 2.宣告方式：boolresult;result=true; 3.可以當作整數用（tru

跟我一起學C++之從C到C++（結構體記憶體對齊）

1.什麼是記憶體對齊 （1）      編譯器為每個“資料單元”按排在某個合適的位置上。 （2）      C、C++語言非常靈活，它允許你干涉“記憶體對齊”。也就是可以人為的設定編譯器的對齊方式。 2.為什麼要對齊 效能原因：在對齊的地址上訪問資料快。如果是位元組對齊方式

跟我一起學.NetCore之配置變更監聽

**前言** 通常程式中配置少不了，配置的修改也避免不了，配置的熱更新為此給應用程式帶來很大的便捷，不用重啟，提高使用者體驗；但往往有時候需要對修改進行審計，也就是需要記錄，有時候也會針對配置修改的時候觸發相關操作，比如說發郵件通知，或是其他業務操作等，遇到這種情況，配置變更監聽的用處就體現出來了，接下來就

跟我一起學.NetCore之選項(Options)核心型別簡介

**前言** .NetCore中提供的選項框架，我把其理解為配置組，主要是將服務中可供配置的項提取出來，封裝成一個型別；從而服務可根據應用場景進行相關配置項的設定來滿足需求，其中使用了依賴注入的形式，使得更加簡單、便捷；另外和配置(Configuration)系統的無縫結合，使得服務更加靈活；而對於Opti

跟我一起學.NetCore之日誌(Log)模型核心

**前言** 魯迅都說：沒有日誌的系統不能上線(魯迅說：這句我沒說過，但是在理)！日誌對於一個系統而言，特別重要，不管是用於事務審計，還是用於系統排錯，還是用於安全追蹤.....都扮演了很重要的角色；之前有很多第三方的日誌框架也很給力，如Log4Net、NLog和Serilog等，在.NetCore中也集成

跟我一起學.NetCore之日誌作用域及第三方日誌框架擴充套件

前言 上一節對日誌的部分核心型別進行簡單的剖析，相信現在再使用日誌的時候，應該大概知道怎麼一回事了，比如記錄器是怎麼來的，是如何將日誌內容寫入到不同目的地的等；當然還有很多細節沒深入講解，抽時間小夥伴們可以去研究研究；廢話不多說，接下來主要舉例演示日誌作用域及第三方日誌框架的擴充套件； 正文 說到日

跟我一起學.NetCore之中介軟體(Middleware)簡介和解析請求管道構建

**前言** 中介軟體(Middleware)對於Asp.NetCore專案來說，不能說重要，而是不能缺少，因為Asp.NetCore的請求管道就是通過一系列的中介軟體組成的；在伺服器接收到請求之後，請求會經過請求管道進行相關的過濾或處理； **正文** 那中介軟體是那路大神？ 會經常聽說，需要註

跟我一起學.NetCore之檔案系統應用及核心淺析

**前言** 在開發過程中，肯定避免不了讀取檔案操作，比如讀取配置檔案、上傳和下載檔案、Web中html、js、css、圖片等靜態資源的訪問；在配置檔案讀取章節中有說到，針對不同配置源資料讀取由對應的IConfigurationProvider進行讀取，其實讀取檔案也是一樣，針對於不同型別(物理檔案、嵌入檔

跟我一起學.NetCore之靜態檔案處理的那些事

**前言** 如今前後端分離開發模式如火如荼，開發職責更加分明（當然前後端一起搞的模式也沒有完全褪去）；而對於每個公司產品實施來說，部署模式會稍有差別，有的會單獨將前端檔案部署為一個站點，有的會將前端檔案和後端站點整合一起部署；通常當專案規模比較大的時候，分開站點部署是不錯的選擇，管理和維護清晰，而對於一些

跟我一起學.NetCore之路由的最佳實現

**前言** 路由，這詞絕對不陌生，不管在前端還是後端都經常提到，而這節不說其他，就聊.NetCore的路由；在之前的Asp.Net MVC 中，路由算是面試時必問的考點，可見其重要性，它的主要作用是對映URL，而不需要關注伺服器的物理檔案結構，提高安全性，同時規範了URL請求，有利於搜尋引擎優化；所以在A

跟我一起學Redis之Redis概述

**背景** 技術的更新迭代，是程式設計師最最最頭大的事，總是在每個網路角落中有感慨聲：學不動啦； ![img](https://i.loli.net/2020/09/23/Zhsc8k6yUJlHEGW.png) 其實新技術並不是憑空而出，而是隨著業務推進、資料驅動、技術積累促使開發者的不斷探索和實踐

跟我一起學.NetCore之Swagger讓前後端不再煩惱及介面自定義

**前言** 隨著前後端分離開發模式的流行，介面對接、聯調成為常事，前端同事會經常問：我需要調哪個介面？這個介面資料格式是啥？條件都傳啥？ 對於一些緊急介面可能會採取溝通對接，然後補文件，其他的都會回一句：看文件。 那難道要一邊開發一邊寫文件嗎？早些年是這樣的，但對於後端同事就很不自在了，程式碼敲的正起勁，

跟我一起學Redis之五種基本型別及其應用場景舉例(幹了6個小時)

**前言** 來啦，老弟？來啦，上一篇就當嘮嘮嗑，接下來就開始進行實操擼命令，計劃是先整體單純說說Redis的各種用法和應用，最後再結合程式碼歸納總結。 Redis預設有16個數據庫(編號為0~15)，預設使用第0個，通過命令select任意切換資料庫，和MySql切換資料庫一個道理；各資料庫之間的資

跟我一起學.NetCore之熟悉的介面許可權驗證不能少(Jwt)

**前言** 許可權管控對於一個系統來說是非常重要的，最熟悉不過的是選單許可權和資料許可權，上一節通過Jwt實現了認證，接下來用它實現介面許可權的驗證，為什麼不是選單許可權呢？對於前後端分離而言，稱其為介面許可權感覺比較符合場景(我是這麼理解的)；資料許可權牽涉到具體業務，這裡就不說啦！ **正文**

跟我一起學.NetCore之MVC過濾器，這篇看完走路可以仰著頭走

**前言** MVC過濾器在之前Asp.Net的時候就已經廣泛使用啦，不管是面試還是工作，總有一個考點或是需求涉及到，可以毫不疑問的說，這個技術點是非常重要的； 在之前參與的面試中，得知很多小夥伴只知道有一兩個過濾器，而對其執行順序瞭解的還是很模糊，少部分小夥伴甚至還沒有使用過。這裡就詳細來說說這塊的內容。