高级特性_广播 | 学习笔记

开发者学堂课程【大数据 Spark 2020版（知识精讲与实战演练）第三阶段：高级特性_广播】学习笔记，与课程紧密联系，让用户快速学习知识。

课程地址：https://developer.aliyun.com/learning/course/689/detail/12032

高级特性_广播

内容简介

一、介绍广播变量的作用和原理

二、API 介绍

三、场景

 

前面了解了全局累加器，接下里了解广播变量，以及在什么场景下会用到广播变量，大致又三个知识点，介绍一下广播变量的作用是什么，API 介绍，场景是什么。

 

一、介绍广播变量的作用和原理

1.引入

一个 execultor 中可以有多个 Task

Class Myclass（

Val field = “hello”

Def dostuff（rdd：RDD（string]):RDD[string] = {

Rdd.map(x => field + x)

//每个分区就是一个 task，这些 task 会被分发到不同的 execultor 中执行

Field 变量会被拷贝给 task，随着 task 一起被分发，execultor 一起反序列化变量，执行相应计算。

Task 如果依赖外部的变量，这样 task 会有单独的外部数位集，会被序列化复制多份

举例

Val map =_

Rdd.flatmap(item =>item + map） // 闭包中引用了 map,map 会被序列化，随着 text 分发到不同的 executor 中执行

例如 task 有十个，map 会被分发十份，copy 十份

*如果有 10 个 task，那么 map 就会被分发十份，就会有是个 map

*但是假如说集群只有2execultor

*无论有多少个 task，最终都会被放在这2 个 execultor 上执行

*一个 executor 代表一个进程，集群需要多少个map？这里只需要两个 task，十个不合适，为解决这个问题，就可以用到广播。

一个 executor 对应了一个 blockmanager，那么它可以把广播放在 executor 对应的 blockmanager 上。

2.原理作用

广播就是在代码执行之前，预先将代码分发到不同的 executor 当中，task 可以直接取出使用，简而言之，广播能够让集群的消耗变少。

二、广播变量的 API

1.使用广播变量

可以通过如下方式创建广播变量

Val b = sc.broadcast（内容）//b 代表广播，内容就是传入的数据

2.如果log 级别为 DEBUG 的时候，会打印如下信息

其实就是将 broadcast 放入 blockmanager 中。

3.创建后可以使用 value 获取数据

b.Value

4.获取数据的时候会打印以下信息

DEBUG blockmanger： getting local block broadcast_0

DEBUG blockmanger: level for block broadcast_0 is storagelevel(disk,memory,deserialized,1 relicas)

//会从 blockmanger 中获取变量

5.广播变量使用完以后，可以使用 unpersist 删除数据

b.unpersist

6.删除数据以后，可以使用 destroy 销毁变量，释放内存空间

b.destroy

7.销毁以后，会打印如下信息

8使用 value 方法的注意点

（1）保证在使用 value 获取广播时，广播没有被清除，被清除后则会爆出以下错误

（2）destroy 方法会移除广播变量，彻底销毁掉，但是如果你试图多次 destroy 广播变量，则会爆出以下错误

Val bc = Sc.broackast（obj)

Rddmap(item =>bc.value)

Val bc = sc.broadcast（ob）

Rdd.map（item = bc.value)//下面会有完整计算方式

 

三、场景

1.创建新的类，broadcast

/**

*资源占用比较大，又是个对应的 value

@test

Def bc1（）：unit = {

//数据，假装这个数据很大，大概一百兆

Val v = map（“spark”-“http://spark.apache.cn”，“Scala”-“http：//www.scala-lang.org”)

Val confing = new sparkconf().setmaster(“local[6]a”).setappname(“bc”)

Val sc = new sparkcontext(config)

//将其中的 spark 和 scala 转为对应的网址

Val r = sc.parallelize(seq(“spark”.”scala”))

Val result = r.map(item => v（item））.Collect（）// task 中引用了 v，v 也要被序列化

Println（result）

/**

*使用广播，大幅度减少 value 的复制

2.应用广播

@test

Def bc2（）：unit = {

//数据，假装这个数据很大，大概一百兆

Val v = map（“spark”-“http://spark.apache.cn”，“Scala”-“http：//www.scala-lang.org”)

//创建广播

val bc = sc.Broadcast（v)   // bc 变成广播

Val confing = new sparkconf().setmaster(“local[6]a”).setappname(“bc”)

Val sc = new sparkcontext(config)

//将其中的 spark 和 scala 转为对应的网址

Val r = sc.parallelize(seq(“spark”.”scala”))

//在算子中使用广播变量代替直接引用集合，这回复制和 executor 一样的数量

//在使用广播之前，复制 map 了 task 数量份

//在使用广播以后，复制次数和 executor 数量一致

Val result = r.map(item => bc.value（item））.Collect（）// task 中引用了 v，v 也要被序列化

Println（result）

//在真实场景下，task 数量会比 executor 数量多得多

运行结果如下