Spark利用广播变量简化大表和小表的join *** 作

两个RDD进行join *** 作(即 rdd1.join(rdd2)) 会导致shuffle ，这是因为join *** 作会对key一致的key-vlaue对进行合并，而** key相同的key-value对不太可能会在同一个partition ， 因此很有可能是需要进行经过网络进行shuffle的，而shuffle会产生许多中间数据（小文件）并涉及到网络传输，这些通常比较耗时，Spark中要尽量避免shuffle 。

优化方法： 将小RDD的数据通过broadcast到每个executor中，各大RDD partition分别和小RDD做join *** 作 。庆败

具体是：在driver端将小RDD转换成数组array并broadcast到做差陆各executor端，然后再各executor task中对各partion的大RDD的key-value对和小rdd的key-value对进行join；由于 每个executor端都有完整的小RDD ，因此小RDD的各partition 不需要shuffle 到RDD的各partition，小RDD广播到大RDD的各partition后， 各partition分别进行join，最后再执行reduce ， 所有分纯顷区的join结果汇总到driver端 。

如有错误，敬请指正！

计算负载主要由 Executors 承担，Driver 主要负责分布式调度，调优空间有限，因此对 Driver 端的配置项我们不作考虑

通过如下参数就可以明确有多少 CPU 资源被划拨给 Spark 用于分布式计算。

spark.cores.max 集群

spark.executor.cores Executor

spark.task.cpus 计算任务

并行度

spark.default.parallelism 并行度

spark.sql.shuffle.partitions 用于明确指定数据关联或聚合 *** 作中 Reduce 端的分区数量。

在平衡 Execution memory 与 Storage memory 的时候，如果 RDD 缓存是刚需，我们就把 spark.memory.storageFraction 调大，并且在应用中优先把缓存灌满，再把计算逻辑应用在缓存数据之上。除此之外，我们还可以同时调整 spark.rdd.compress 和 spark.memory.storageFraction 来缓和 Full GC 的冲击

spark.local.dir 这个配置项，这个参数允许开发者设置磁盘目录，该目录用于存储 RDD cache 落盘数据块和 Shuffle 中间文件。如果你的经费比较充胡谈裕，有条件在计算节点中配备足量的 SSD 存储，甚至是更多的内存资源，完全可以把 SSD 上的文件系统目录，或是内存文件系统添加到 spark.local.dir 配置项中去，从而提供更好的 I/O 性能。

Configuration - Spark 3.2.1

自 1.6 版本之后，Spark 统一采用 Sort shuffle manager 来管理 Shuffle *** 作，在 Sort shuffle manager 的管理机制下，无论计算结果本身是否需要排序，Shuffle 计算过程在 Map 阶段和 Reduce 阶段都会引入排序 *** 作。

在不需要聚合，也不需要排序的计算场景中，我们就可以通过设置 spark.shuffle.sort.bypassMergeThreshold 的参数，来改变 Reduce 端的并行度（默认乎凯值是 200）。当 Reduce 端的分区数小于这个设置值的时候，我们就能避免 Shuffle 在计算过程引入排序。

Spark SQL 下面的配置项还是蛮多的，其中对执行性能贡献最大的，当属 AQE（Adaptive query execution，自适应查询引擎）引入的那 3 个特性了，也就是自动分区合并、自动数据倾斜处理和 Join 策略调整。

AQE 事先并不判断哪些分区足够小，而是按照分区编号进行扫描，当扫描量超过“目标尺寸”时，就合并一次,那么，“目标尺寸”由什么来决定的呢？Spark 提供了两个配置项来共同决定分区合并的“目裤顷碰标尺寸”,分区合并的目标尺寸取 advisoryPartitionSizeInBytes 与 partitionSize (每个分区的平均大小)之间的最小值。

我们来举个例子。假设，Shuffle 过后数据大小为 20GB，minPartitionNum 设置为 200，反推过来，每个分区的尺寸就是 20GB / 200 = 100MB。再假设，advisoryPartitionSizeInBytes 设置为 200MB，最终的目标分区尺寸就是取（100MB，200MB）之间的最小值，也就是 100MB。因此你看，并不是你指定了 advisoryPartitionSizeInBytes 是多少

首先，分区尺寸必须要大于 spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes 参数的设定值，才有可能被判定为倾斜分区。然后，AQE 统计所有数据分区大小并排序，取中位数作为放大基数，尺寸大于中位数一定倍数的分区会被判定为倾斜分区，中位数的放大倍数也是由参数 spark.sql.adaptive.skewJoin.skewedPartitionFactor(默认值是 5 倍) 控制。

实际上指的是，把会引入 Shuffle 的 Join 方式，如 Hash Join、Sort Merge Join，“降级”（Demote）为 Broadcast Join。

在 Spark 发布 AQE 之前，开发者可以利用 spark.sql.autoBroadcastJoinThreshold 配置项对数据关联 *** 作进行主动降级。这个参数的默认值是 10MB，参与 Join 的两张表中只要有一张数据表的尺寸小于 10MB

不过，autoBroadcastJoinThreshold 这个参数虽然好用，但是有两个让人头疼的短板。一是可靠性较差。尽管开发者明确设置了广播阈值，而且小表数据量在阈值以内，但 Spark 对小表尺寸的误判时有发生，导致 Broadcast Join 降级失败。二来，预先设置广播阈值是一种静态的优化机制，它没有办法在运行时动态对数据关联进行降级调整。

AQE 很好地解决了这两个头疼的问题。首先，AQE 的 Join 策略调整是一种动态优化机制，对于刚才的两张大表，AQE 会在数据表完成过滤 *** 作之后动态计算剩余数据量，当数据量满足广播条件时，AQE 会重新调整逻辑执行计划，在新的逻辑计划中把 Shuffle Joins 降级为 Broadcast Join。再者，运行时的数据量估算要比编译时准确得多，因此 AQE 的动态 Join 策略调整相比静态优化会更可靠、更稳定。

每个 Map Task 生成的数据文件，都包含所有 Reduce Task 所需的部分数据。因此，任何一个 Reduce Task 要想完成计算，必须先从所有 Map Task 的中间文件里去拉取属于自己的那部分数据。索引文件正是用于帮助判定哪部分数据属于哪个 Reduce Task。Reduce Task 通过网络拉取中间文件的过程，实际上就是不同 Stages 之间数据分发的过程。

显然，Shuffle 中数据分发的网络开销，会随着 Map Task 与 Reduce Task 的线性增长，呈指数级爆炸。

Shuffle Joins

第一步就是对参与关联的左右表分别进行 Shuffle，Shuffle 的分区规则是先对 Join keys 计算哈希值，再把哈希值对分区数取模。Shuffle 完成之后，第二步就是在同一个 Executors 内，Reduce task 就可以对 userID 一致的记录进行关联 *** 作。

Broadcast Join

使用广播阈值配置项让 Spark 优先选择 Broadcast Joins 的关键，就是要确保至少有一张表的存储尺寸小于广播阈值（数据表在磁盘上的存储大小，同一份数据在内存中的存储大小往往会比磁盘中的存储大小膨胀数倍）

Spark 将内存分成了 Execution Memory 和 Storage Memory 两类，分别用于分布式任务执行和 RDD 缓存。其中，RDD 缓存虽然最终占用的是 Storage Memory，但在 RDD 展开（Unroll）之前，计算任务消耗的还是 Execution Memory。因此，Spark 中 CPU 与内存的平衡，其实就是 CPU 与执行内存之间的协同与配比。

并行度指的是为了实现分布式计算，分布式数据集被划分出来的份数。并行度明确了数据划分的粒度：并行度越高，数据的粒度越细，数据分片越多，数据越分散。并行度可以通过两个参数来设置，分别是 spark.default.parallelism 和 spark.sql.shuffle.partitions。前者用于设置 RDD 的默认并行度，后者在 Spark SQL 开发框架下，指定了 Shuffle Reduce 阶段默认的并行度。并发度呢？Executor 的线程池大小由参数 spark.executor.cores 决定，每个任务在执行期间需要消耗的线程数由 spark.task.cpus 配置项给定。两者相除得到的商就是并发度，也就是同一时间内，一个 Executor 内部可以同时运行的最大任务数量。又因为，spark.task.cpus 默认数值为 1，并且通常不需要调整，所以，并发度基本由 spark.executor.cores 参数敲定。就 Executor 的线程池来说，尽管线程本身可以复用，但每个线程在同一时间只能计算一个任务，每个任务负责处理一个数据分片。因此，在运行时，线程、任务与分区是一一对应的关系。

对于 User Memory 内存区域来说，使用 空间去重复存储同样的数据，本身就是降低了内存的利用率

对于存储级别来说，实际开发中最常用到的有两个，MEMORY_ONLY 和 MEMORY_AND_DISK，它们分别是 RDD 缓存和 DataFrame 缓存的默认存储级别。对于缓存计算来说，它分为 3 个步骤，第一步是 Unroll，把 RDD 数据分片的 Iterator 物化为对象值，第二步是 Transfer，把对象值封装为 MemoryEntry，第三步是把 BlockId、MemoryEntry 价值对注册到 LinkedHashMap 数据结构。另外，当数据缓存需求远大于 Storage Memory 区域的空间供给时，Spark 利用 LinkedHashMap 数据结构提供的特性，会遵循 LRU 和兔子不吃窝边草这两个基本原则来清除内存空间：LRU：按照元素的访问顺序，优先清除那些“最近最少访问”的 BlockId、MemoryEntry 键值对兔子不吃窝边草：在清除的过程中，同属一个 RDD 的 MemoryEntry 拥有“赦免权”

PROCESS_LOCAL：任务与数据同在一个 JVM 进程中

NODE_LOCAL：任务与数据同在一个计算节点，数据可能在磁盘上或是另一个 JVM 进程中

RACK_LOCAL：任务与数据不在同一节点，但在同一个物理机架上

ANY：任务与数据是跨机架、甚至是跨 DC（Data Center，数据中心）的关系访问数据源是否会引入网络开销，取决于任务与数据的本地性关系，也就是任务的本地性级别

Shuffle 作为大多数计算场景的“性能瓶颈担当”，确实是网络开销的罪魁祸首。根据“能省则省”的开发原则，我们自然要想尽办法去避免 Shuffle。

在数据通过网络分发之前，我们可以利用 Kryo Serializer 序列化器，提升序列化字节的存储效率，从而有效降低在网络中分发的数据量，整体上减少网络开销。

不论是Hive还是Spark SQL在使用过程中都可能会遇到小文件过多的问题。小文件过多最直接的表现是任务执行时间长，查看Spark log会发现大量的数据移动的日志。我们可以查看log中展现的日志信息，去对应的路径下查看文件的大小和个数。租敬稿

通过上述命令可以查看文件的个数以及大小。count查看出的文件大小单位是B，需要转换为MB。

在spark官方的推荐文档中，parquet格式的文件推荐大小是128MB，小于该大小的均可以称之为小文件，在实际的工作，往往小文件的大小仅仅为几KB，表现为，可能文件大小为几百MB，但是文件个数可能到达了几十万个。一般来说，我们可以通过简单相除获得文件的平均大小，如果文件数目不多，我们也可以通过下述命令获得每个文件的大小。

1.任务执行时间长

2.真实的文件大小独占一个数据存储块，存放到DataNode节点中。同时 DataNode一般默认存三份副本，以保障数据安全。同时该文件所存放的位置也写入到NameNode的内存中，如果有Secondary NameNode高可用节点，也可同时复制一份过去。NameNode的内存数据将会存放到硬盘中，如果HDFS发生重弊孝启，将产生较长时间的元数据从硬盘读到内存的过程。

3.不论在Hive还是在Spark中，每一个存储块都对应一个Map程序，一个Map呈现就需要一个JVM，启动一个JVM去读取或者写小文件是吃力不讨好的行为。在实际的生产中，为了更好的管理集群资源，一般会要求程序执行时限制Executor数量和每个Executor的核心数量，需要频繁创建Executor来读取写入。

5.影响磁盘寻址时间

小文件合并，本质上就是通过某种 *** 作，将一系列小文件合并成大文件。我们知道，以MapReduce为代表的大数据系统，都习惯用K-V键值对的形式来处理文件，最后文件落盘，也是一个reduce对应一个输出文件。所以直观上，我们可以减少reduce数量，达到减少文件数量的目的。

从Map到Reduce需要一个Shuffle过程，所以我们将小文件合并理解为通过一个Shuffle，合并小文件成一个大文件。基于这样的思想，我们的策略可以稿耐分为两类：一类是原来的计算已经有Shuffle了，那么我们可以认为控制输出文件的数量；二类是强制触发Shuffle，进行小文件合并。

1-设置参数 (一般用于Hive)

2-distribute by rand()

往动态分区插入数据时，在已经写好的SQL末尾加上distribute by rand()

该算子只是起到打散的效果，但是我们还要设置文件的大小，以免打散后仍然有小文件。

表示每个reduce的大小，Hive可以数据总量，得到reduce个数，假设hive认为会有10个reduce,那么，这里rand()则会为 x % 10

3-group by

我们知道，group by算子会触发Shuffle，因此只要我们设置好Shuffle时的文件个数就好，在Spark SQL中，我们可以设置partition个数，因为一个partition会对应一个文件。

上述的 *** 作，会触发shuffle，因此我们再设置partition个数。

则表示，shuffle后，只会产生10个partition.

4-repartition()

5-coalesce()

需要注意的是，4和5都是spark 2.4以及以后才会支持的。

---