天天看点

Spark Broadcast内幕解密(简约版)

一:Broadcast彻底解析

1,Broadcast就是将数据从一个节点发送到其他的节点上;例如Driver上有一张表,而Executor中的每个并行的Task(100万个Task)都要查询这张表,那我们通过Broadcast方式就只需要往每个Executor把这张表发送一次就行了,Executor中的每个Task查询这张唯一的表,而不是每次执行的时候都从Driver获得这张表!

2,这就好像ServletContext的具体作用,只是Broadcast是分布式的共享数据,默认情况下只要程序在运行Broadcast变量就会存在,因为Broadcast子底层是通过BroadcastManager管理的!但是你可以手动指定或者配置具体周期来销毁Broadcast变量!

3,Broadcast一般用于处理共享配置文件、通用的DataSet、常用的数据结构等等;但是不适合存放太大的数据在Broadcast,Broadcast不会内存溢出,因为其数据的保存的StoreLevel是MEMORY_AND_DISK的方式;虽然如此,我们也不可以放入太大的数据在Broadcast中,因为网络IO和可能的单点压力会非常大!

4,Broadcast变量是只读变量,最为轻松保持了数据的一致性!

5,Broadcast的使用:

* {{{
* scala> val broadcastVar = sc.broadcast(Array(1, 2, 3))
* broadcastVar: org.apache.spark.broadcast.Broadcast[Array[Int]] = Broadcast(0)
*
* scala> broadcastVar.value
* res0: Array[Int] = Array(1, 2, 3)
* }}}      

6,HttpBroadcast方式的Broadcast,最开始的时候数据放在Driver的本地文件系统中,Driver在本地会创建一个文件夹来存放Broadcast中的data,然后启动HttpServer来访问文件夹中的数据,同时写入到BlockManager(StorageLevel是MEMORY_AND_DISK)中获得BlockId(BroadcastBlockId),当Executor中的Task要访问Broadcast变量的时候,会向Driver通过HttpServer来访问数据,然后会在Executor中的BlockManager中注册该Broadcast中的数据,这样后要的Task需要访问Broadcast的变量的时候会首先查询BlockManager中有没有该数据,如果有就直接使用;

7,BroadcastManager是用来管理Broadcast,该实例对象是在SparkContext创建SparkEnv的时候创建的:

// Called by SparkContext or Executor before using Broadcast
private def initialize() {
  synchronized {
    if (!initialized) {
      val broadcastFactoryClass =
        conf.get("spark.broadcast.factory", "org.apache.spark.broadcast.TorrentBroadcastFactory")

      broadcastFactory =
        Utils.classForName(broadcastFactoryClass).newInstance.asInstanceOf[BroadcastFactory]

      // Initialize appropriate BroadcastFactory and BroadcastObject
      broadcastFactory.initialize(isDriver, conf, securityManager)

      initialized = true
    }
  }
}      

在实例化BlockcastManager的时候会创建BlockcastFactory工厂来构建具体实际的Brockcast类型,默认情况下是TorrentBroadcastFactory;

继续阅读