一、问题
两个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端。
利用广播变量简化大表和小表的join操作
三、业务代码
import org.apache.spark.sql.SparkSession
object BigRDDJoinSmallRDD {
def main(args: Array[String]): Unit = {
val sparkSession = SparkSession.builder().master("local[3]").appName("BigRDD Join SmallRDD").getOrCreate()
val sc = sparkSession.sparkContext
val list1 = List(("jame",23), ("wade",3), ("kobe",24))
val list2 = List(("jame", 13), ("wade",6), ("kobe",16))
val bigRDD = sc.makeRDD(list1)
val smallRDD = sc.makeRDD(list2)
println(bigRDD.getNumPartitions)
println(smallRDD.getNumPartitions)
// driver端rdd不broadcast广播smallRDD到各executor，RDD不能被broadcast，需要转换成数组array
val smallRDDB= sc.broadcast(smallRDD.collect())
val joinedRDD = bigRDD.mapPartitions(partition => {
val smallRDDBV = smallRDDB.value // 各个executor端的task读取广播value
partition.map(element => {
//println(joinUtil(element, smallRDDBV))
joinUtil(element, smallRDDBV)
})
})
joinedRDD.foreach(x => println(x))
}
/**
* join操作：对两个rdd中的相同key的value1和value2进行聚合，即(key,value1).join(key,value2)得到(key,(value1, vlaue2))
* 如果bigRDDEle的key和smallRDD的某个key一致，那么返回(key,(value1, vlaue2))
* 该方法会在各executor的task上执行
* */
def joinUtil(bigRDDEle:(String,Int), smallRDD: Array[(String, Int)]): (String, (Int,Int)) = {
var joinEle:(String, (Int, Int)) = null
// 遍历数组smallRDD
smallRDD.foreach(smallRDDEle => {
if(smallRDDEle._1.equals(bigRDDEle._1)){
// 如果bigRDD中某个元素的key和数组smallRDD的key一致，返回join结果
joinEle = (bigRDDEle._1, (bigRDDEle._2, smallRDDEle._2))
}
})
joinEle
}
}
如有错误，敬请指正！
