spark 开启 mapjoin 参数

Spark 中开启 MapJoin 需要设置一个阈值参数，当参与 Join 的两个表中较小的那个表的大小小于这个阈值时，Spark 会自动采用 MapJoin 的方式来执行 Join 操作。MapJoin 通常会将小表广播到各个节点上，然后在每个节点上对大表的每个分区执行 Join 操作，这样可以避免 Shuffle 的开销，提高 Join 操作的效率。

在 Spark SQL 中，你可以通过 `spark.sql.autoBroadcastJoinThreshold` 参数来设置这个阈值。这个值的单位是字节，默认值是 10MB。如果你想要改变这个阈值，可以在配置 SparkSession 的时候进行设置：

val spark = SparkSession.builder()
    .appName("SparkExample")
    .config("spark.sql.autoBroadcastJoinThreshold", "20971520") // 设置为 20MB
    .getOrCreate()

需要注意的是，虽然 MapJoin 可以提高性能，但是由于需要将小表广播到各个节点上，因此如果小表的实际大小超过了集群的内存容量，就可能引发内存溢出的问题。因此，在实际应用中需要根据集群的实际内存情况来合理设置这个阈值。

相关问题

spark 数据倾斜

### 解决 Spark 数据倾斜问题的最佳实践

#### 设置合理的 Shuffle Partition 数量
为了防止由于分区数量不足而导致的数据倾斜，在配置 Spark 应用程序时应适当增加 `spark.sql.shuffle.partitions` 参数的值。此参数决定了 shuffle 类操作（如 group by, join）产生的分区数目，默认情况下为200，这可能不足以应对大规模数据集的情况[^5]。

// 增加shuffle partitions的数量以减少单个task的压力
val sparkSession = SparkSession.builder()
  .appName("DataSkewExample")
  .config("spark.sql.shuffle.partitions", "500") // 调整至更合适的数值
  .getOrCreate()

#### 使用广播变量优化 Join 操作
当参与 join 的两个 RDD 或 DataFrame 中有一个较小而另一个较大时，可以通过广播机制将小表加载到内存中作为广播变量传递给每一个工作节点上的任务去执行 join 操作。这样不仅可以提高效率还能缓解因 key 分布不均引起的数据倾斜现象[^3]。

import org.apache.spark.broadcast.Broadcast

// 对于小表bdf创建广播变量
val broadcastVar: Broadcast[DataFrame] = sparkSession.sparkContext.broadcast(bdf)

// 在大表adf上应用mapPartitions函数完成join逻辑
val resultDf = adf.mapPartitions(partition => {
    val localBDF = broadcastVar.value
    partition.flatMap(rowA => {
        val matchingRowsInB = localBDF.filter(_.key == rowA.key).collect() // 这里假设存在名为'key'的列用于关联两表
        matchingRowsInB.map(rowB => (rowA, rowB))
    })
})

#### 合并少量的小分区
如果某些阶段结束后出现了大量空闲或几乎为空的任务，则意味着这些任务对应的输入数据很少甚至不存在。此时应该考虑合并那些规模很小的分区成更大的几个分区来进行后续计算，从而降低资源浪费并改善性能表现[^4]。

SET spark.sql.adaptive.enabled=true;
-- 开启自适应查询计划功能后，Spark会自动尝试调整分区大小

#### 随机前缀法处理极端倾斜情况
针对特别严重且难以通过上述方法完全消除影响的关键字分布模式，可以在原有 key 上附加随机生成的部分位数作为新的组合键进行重新分配，使得原本集中在少数几个特定位置的数据能够更加均匀地散布开来被不同的 worker 处理。

def addRandomPrefix(key: String): String = s"${new java.util.Random().nextInt(10)}-$key"

rdd.map{case (k,v) => (addRandomPrefix(k), v)}
   .reduceByKey(_ + _)
   .map{case (prefixKey,value) => prefixKey.split("-")(1)->value}