Tag - spark

spark    2019-04-22 02:54:13    15    0    0

1. 使用场景

在spark开发过程中,最常见的用到广播的场景有两个:

  1. 共享第三方存储客户端和连接池(Redis、HBase等):用于读写第三方存储,如果不使用广播可能造成频繁的创建连接,影响性能(即使基于分区创建连接,在分区数较多的情况下也不能避免)。
  2. 共享只读数据:多次使用的只读数据可以使用广播发送到每个节点,避免重复传输,起到性能优化的作用。

2. 实现方式

广播的实现分为几个步骤:

  1. 序列化:由Driver将数据序列化,生成chunks保存在Driver的BlockManager中。
  2. 传输读取:Executor在使用广播变量时首先尝试从本地BlockManager中获取数据,如果获取失败则从Driver或其他Executor的BlockManager获取,最终在首次使用时将数据反序列化成对象。

传输实现有两种方式:HttpBroadcast所有Executor都从Driver获取广播变量、TorrentBroadcastExecutor之间也可以传输广播变量,避免Driver性能瓶颈。

3.示例

使用广播变量的示例网上有很多,这里只列举一个共享第三方存储客户端和连接池的示例。很多初写spark的同事问完怎么共享Redis连接池后又会问怎么广播的时候报不能序列化的异常NotSerializableException,这里强调下,广播的第一步就是将对象序列化,所以类必须继承Seriable接口,但是很多第三方存储的客户端没有实现该接口,就导致这个问题。

这里我们为了避免这个问题,只需要使用懒加载即可,即在Driver序列化的时候对象是可序列化的就能满足要求:

  1. import org.apache.spark.{SparkConf, SparkContext}
  2. import redis.clients.jedis.{HostAndPort, JedisCluster}
  3. //使用RedisSink来封装Redis连接池,避免直接广播JedisCluster
  4. class RedisSink extends Serializable {
  5. //使用lazy修饰的变量,序列化时不会初始化此变量
  6. lazy val clu