大数据 - Spark系列《九》- 广播变量

  def main(args: Array[String]): Unit = {
    val sc: SparkContext = SparkUtil.getSc
    val rdd1: RDD[(String, Int)] = sc.parallelize(List(("a", 1), ("b", 2), ("c", 3)))
    val rdd2: RDD[(String, Int)] = sc.parallelize(List(("a", 11), ("b", 22), ("c", 33)))
    // 数据在进行join的时候会出现shuffle , 我们在编程的时候尽量避免shuffle的产生
    val resRDD1: RDD[(String, (Int, Int))] = rdd1.join(rdd2)

    // 使用这样的方式来实现join逻辑, 这里调用的是map函数是没有shuffle产生的
     // 1 将数据收集到driver端
    val mp: Map[String, Int] = rdd2.collect().toMap
    // map函数的逻辑分布式执行在不同的Task中 , 那么我们的mp数据在每个Task中都存在
    val resRDD2: RDD[(String, (Int, Int))] = rdd1.map {
      case (k, v) => {
        val i: Int = mp.getOrElse(k, -1)
        (k, (v, i))
      }
    }
    resRDD2.foreach(println)
    sc.stop()

4. 🥙使用广播变量

package com.doit.day0217
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import scala.io.Source
/**
 * @日期: 2024/2/19
 * @Author: Wang NaPao
 * @Blog: https://blog.csdn.net/weixin_40968325?spm=1018.2226.3001.5343
 * @Tips: 和我一起学习吧
 * @Description:
 */


object Test09 {
  def main(args: Array[String]): Unit = {
    // 创建SparkConf对象，并设置应用程序名称和运行模式
    val conf = new SparkConf()
      .setAppName("Starting...") // 设置应用程序名称
      .setMaster("local[*]") // 设置运行模式为本地模式

    // 创建SparkContext对象，并传入SparkConf对象
    val sc = new SparkContext(conf)

    // 加载用户数据，将其转换为Map
    val user = Source.fromFile("data/join/user.txt")
    val userMap: Map[String, String] = user.getLines().map(line => {
      val arr = line.split(",")
      (arr(0), arr(1))
    }).toMap // 将用户数据转换为Map，存储在内存中 一般数据集控制大小为1G以内

    //1)--------------------  将map集合变成广播变量
    val bc = sc.broadcast(userMap)
    // 加载订单数据
    val orders = sc.textFile("data/join/orders.txt")

    // 使用闭包变量userMap避免shuffle，将用户名直接关联到订单数据中
    val rdd2 = orders.map(iter => {
      val arr2 = iter.split(",")
      //val name = userMap.getOrElse(arr2(4), "unknown") // 使用闭包变量userMap关联订单数据中的用户ID
      // 2)--------------------  从广播变量中取出数据
      val name = bc.value.getOrElse(arr2(4), "unknown")
      (arr2(0), arr2(1), arr2(2), arr2(3), arr2(4), name)
    })

    // 打印处理后的订单数据
    rdd2.foreach(println)

    // 关闭SparkContext对象
    sc.stop()
  }
}

5. 销毁广播变量

从driver端发送一个RemoveBroadcast消息。在Executor上的BlockManager服务接收该消息，就会把广播变量从BlockManager中删除。若removeFromDriver设置成True，还会从Driver删除该变量的数据。

// 销毁广播变量
bc.unpersist()

9.5 广播变量分发读取机制

1. 创建原理

广播变量的创建发生在Driver端，如图所示，当调用SparkContext#broadcast来创建广播变量时，会把该变量的数据切分成多个数据块，保存到driver端的BlockManger中，使用的存储级别是：MEMORY_AND_DISK_SER。

所以，广播变量的读取也是懒加载的，只有在Executor端需要获取广播变量时才会去获取。此时广播变量的数据只在Driver端存在。

2. 读取原理

1）第1步（红色线1）：首先从本Executor的BlockManager中（或本机的其他executor）读取广播变量的数据，若存在就直接获取，并返回。若不存在，则执行2或3。

2）第2步（红色线2）：从远端获取数据。先从同一个机架(rack)的主机的Executor端获取。若不能从其他Executor中获取广播变量，则会直接从Driver端获取。

3）第3步（红色线3）：从Driver端获取广播变量的状态和位置信息（由于所有的BlockManager slave端都会向Master端汇报数据块状态）。

从以上获取流程可以看出，在执行spark应用时，只要有一个worker节点的Executor从Driver端获取到了广播变量的数据，则其他的Executor就不需要从Driver端获取了。

BT协议

当某个Executor上的某个数据块被删除，可以从其他Executor直接获取该数据块，然后把数据块保存到自己的Executor的BlockManager中。

Executor4中的任务需要使用广播变量，但它只有该变量的b4数据块。此时，它首先从同主机（worker2节点）的中获取数据，获取到数据块；然后分别从不同主机的Executor1和Executor2中读取数据块。此时，Executor4就获取到变量b的全部数据块了，然后把这些数据块在自己的BlockManager中保存一份。此时，其他Executor就可以从Executor4中读取数据了。

当完成这些操作后，各个Executor端的BlockManager（slave端）会向Driver端的BlockManager（master端）汇报数据块的状态。

原文地址:https://blog.csdn.net/weixin_40968325/article/details/136198917 本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：https://www.suanlizi.com/kf/1760493111466921984.html 如若内容造成侵权/违法违规/事实不符，请联系《酸梨子》网邮箱：1419361763@qq.com进行投诉反馈，一经查实，立即删除！

阅读全部