spark join

在大数据处理场景中，多表Join是很是常见的一类运算。为了便于求解，一般会将多表join问题转为多个两表链接问题。两表Join的实现算法很是多，通常咱们会根据两表的数据特色选取不一样的join算法，其中，最经常使用的两个算法是map-side join和reduce-side join。本文将介绍如何在apache spark中实现这两种算法。

（1）Map-side Join

Map-side Join使用场景是一个大表和一个小表的链接操做，其中，“小表”是指文件足够小，能够加载到内存中。该算法能够将join算子执行在Map端，无需经历shuffle和reduce等阶段，所以效率很是高。

在Hadoop MapReduce中， map-side join是借助DistributedCache实现的。DistributedCache能够帮咱们将小文件分发到各个节点的Task工做目录下，这样，咱们只需在程序中将文件加载到内存中（好比保存到Map数据结构中），而后借助Mapper的迭代机制，遍历另外一个大表中的每一条记录，并查找是否在小表中，若是在则输出，不然跳过。

在Apache Spark中，一样存在相似于DistributedCache的功能，称为“广播变量”（Broadcast variable）。其实现原理与DistributedCache很是相似，但提供了更多的数据/文件广播算法，包括高效的P2P算法，该算法在节点数目很是多的场景下，效率远远好于DistributedCache这种基于HDFS共享存储的方式，具体比较可参考“Performance and Scalability of Broadcast in Spark”。使用MapReduce DistributedCache时，用户须要显示地使用File API编写程序从本地读取小表数据，而Spark则不用，它借助Scala语言强大的函数闭包特性，能够隐藏数据/文件广播过程，让用户编写程序更加简单。

假设两个文件，一小一大，且格式相似为：

Key,value,value

Key,value,value

则利用Spark实现map-side的算法以下：

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23

var table 1 = sc.textFile(args( 1 )) var table 2 = sc.textFile(args( 2 ))   // table1 is smaller, so broadcast it as a map<String, String> var pairs = table 1 .map { x = >    var pos = x.indexOf( ',' )    (x.substring( 0 , pos), x.substring(pos + 1 )) }.collectAsMap var broadCastMap = sc.broadcast(pairs) //save table1 as map, and broadcast it   // table2 join table1 in map side var result = table 2 .map { x = >    var pos = x.indexOf( ',' )    (x.substring( 0 , pos), x.substring(pos + 1 )) }.mapPartitions({ iter = >    var m = broadCastMap.value    for {      (key, value) <- iter      if (m.contains(key))    } yield (key, (value, m.get(key).getOrElse( "" ))) })   result.saveAsTextFile(args( 3 )) //save result to local file or HDFS

（2）Reduce-side Join

当两个文件/目录中的数据很是大，难以将某一个存放到内存中时，Reduce-side Join是一种解决思路。该算法须要经过Map和Reduce两个阶段完成，在Map阶段，将key相同的记录划分给同一个Reduce Task（需标记每条记录的来源，便于在Reduce阶段合并），在Reduce阶段，对key相同的进行合并。

Spark提供了Join算子，能够直接经过该算子实现reduce-side join，但要求RDD中的记录必须是pair，即RDD[KEY, VALUE]，一样前一个例利用Reduce-side join实现以下：

1 2 3 4 5 6 7 8 9 10 11 12 13 14

var table 1 = sc.textFile(args( 1 )) var table 2 = sc.textFile(args( 2 ))   var pairs = table 1 .map{x = >    var pos = x.indexOf( ',' )    (x.substring( 0 , pos), x.substring(pos + 1 )) }   var result = table 2 .map{x = >    var pos = x.indexOf( ',' )    (x.substring( 0 , pos), x.substring(pos + 1 )) }.join(pairs)   result.saveAsTextFile(args( 3 ))

（3）总结

本文介绍了Spark中map-side join和reduce-side join的编程思路，但愿对你们有借鉴意义。但须要注意的是，在使用这两种算法处理较大规模的数据时，一般须要对多个参数进行调优，不然可能会产生OOM问题。一般须要调优的相关参数包括，map端数据输出buffer大小，reduce端数据分组方法（基于map仍是基于sort），等等。

（4）两个问题

问题1：若是在map-side join中，不使用如下语句对文件1进行广播，

var broadCastMap = sc.broadcast(pairs)
也能够在后面程序中直接使用变量pairs存储的数据进行join，这两种方式有什么异同，性能会有何不一样？
问题2：将map-side join中的如下语句：123456mapPartitions({ iter =>   var m = broadCastMap.value   for{     (key, value) <- iter     if(m.contains(key))   } yield (key, (value, m.get(key).getOrElse("")))改成：1234var m = broadCastMap.value //这一句放在var table2 = sc.textFile(args(2))后面 map {case (key, value) =>   if(m.contains(key)) (key, (value, m.get(key).getOrElse(""))) }最终结果是有问题的，为何？  本文两个示例程序能够从百度网盘上下载，地址为Spark-Join-Exmaple。