上篇文章演示了在 Spark 开发时，怎么样用 Redis「连接单例」模式引入「外部数据源」的解决方案。

「连接单例」最大的优点，可以保护外部数据源，也就是 Redis 的连接，不至于被一些蹩脚的代码给干爆掉，从而影响 Redis 的正常对外服务。

在上篇文末，我出了个两个思考题，其中第一个是问题是：如果要把上次那个只能在「单线程」方式下运行的 Spark 程序，提交到集群，该怎么设置提交参数？

看评论区已经有小伙伴给出了答案，那就是：可以设置多个 executor，但是每个 executor 只给分配一个 CPU 核心。

比如提交到 yarn 集群，可以这么来设置：

--executor-cores 1 --num-executors 3

这样就可以既保证这个 Spark 任务是分布式运行的，但同时，又保护了「连接单例」的可用性，因为它确保了「程序每个 JVM 内部，是用单线程工作，从而避免了对连接对象的争抢」。

但是，上文也说了，这种方法虽然能解决问题，但它却并不十分体面。

那么今天这篇文章，就来回答上篇文章留下的第二个问题，用一种更体面的方式，来实现对这个 Redis 外部数据源的引入——广播变量。

0. 解读广播变量

估计很多玩过 Spark 的同学都知道什么是「广播变量」，以及上上篇文章聊到的「累加器」，但是能够深入理解，并能把它们灵活玩转的人，一定不多。

先说它们的相同点：

1. 都是从 Driver 端，给序列化到 executor 端的对象；

2. 因为第1条的缘故，所以都不适合承载很大的数据量；

3. 因为需要实现网络传输，所以两者必须都可以被序列化，之前听讨论群里有同学给其他人提建议，说可以把数据库连接作为广播变量，给传到 executor 端，我心想「这哥们一定只是口嗨，因为数据库连接，它不能被序列化，广播个嘚」。

再说它们的不同点：

1. 「累加器」是线程安全的，除了可读，还可以对它进行累加操作，并且保证「线程/进程间即时可见」，考虑的因素比较多。

而「广播变量」仅仅只是保证所有 executor 可读(一次性读取，不能更新)，没那么多花里胡哨的功能，技术含量更低一些；

2. 「累加器」在使用的时候，我们不需要额外考虑往里扔的对象，能不能被序列化的问题，因为但凡你是个正常人，最多只会往里面扔数据，而数据本身，绝大部分是可以被序列化的。

但是在用「广播变量」的时候，你就得注意了，不是想当然，以为万物皆可广播，得具备可以被序列化的潜质，比如上面提到的数据库连接，它就不行。

1. 为什么用广播变量

Redis 中的黑名单数据一般来说，不会太大，正常来说，维护个几百万，甚至上千万的数据量都没什么问题(我就这么干过)，跟广播变量不宜使用过大数据集的要求相符。

而且广播变量数据，满足「一次性获得，多次读取」的特性，跟我们的业务中，要对数据进行“碰撞”的场景吻合。

对于 Redis 来说，可以只需要用到「一次数据库连接」，避免了对 Redis 的连接管理问题。

2. 代码实现

package com.anryg.bigdata.test.variable

import java.util
import com.anryg.bigdata.RedisClientUtils
import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession

/**
  * @DESC: 测试 Spark 用广播变量读取外部数据源
  * @Auther: Anryg
  * @Date: 2024/7/4 17:36
  */
object SparkReadExternalDataTest02 {

    /**
      * @DESC: 从Redis读取黑名单数据
      * */
    private def getBlackDomain(): util.Set[String] ={
        val blackDomain = RedisClientUtils.getSetResult(RedisClientUtils.getSingleRedisClient, /**用Jedis连接单例来获取黑名单数据*/
            5, /**Redis的db编号*/
            "black_domain"  /**对应黑名单的key*/
        )
        blackDomain
    }

    def main(args: Array[String]): Unit = {
        Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
        val sparkConf = new SparkConf().setAppName("SparkReadExternalDataTest01")
            .setMaster("local[3]")
        val spark = SparkSession.builder().config(sparkConf).getOrCreate()

        val broadcast = spark.sparkContext.broadcast(getBlackDomain()) /**将获取的黑名单数据广播出去*/

        /**构造数据源(主数据)*/
        val seq = Seq("baidu.com",
            "qq.com",
            "xiaomi.com",
            "12345.com",
            "abc.com",
            "def.net",
            "xxx.com",
            "google.com",
            "sina.com",
            "163.com",
            "yyy.net",
            "360.com",
            "cctv.cn",
            "12306.com")

        spark.sparkContext.makeRDD(seq) /**生成 RDD 对象*/
            .foreachPartition(iter => {
                val blackDomain = broadcast.value /**从广播变量中获取黑名单数据*/
                while (iter.hasNext){
                    val domain = iter.next()
                    if (!blackDomain.contains(domain)) println(domain)  /**碰撞后，输出合法域名*/
                }
        })
    }
}