在Spark中实现map-side join和reduce-side join_spark mapjoin

以下说的都是二表Join，多表join则可以通过转化为多个二表join来实现。

Map-side Join

如果要join的表中一个是大表，一个是小表（小到可以加载到内存中），就可以采用该算法。该算法可以将join算子执行在Map端，无需经历shuffle和reduce等阶段，因此效率非常高。

类似于Hadoop MapReduce中采用DistributedCache技术来实现map-side join，Spark中有一个Broadcast variable，与DistributedCache非常类似，但是它采用了BitTorrent（就是下载电影的那个BT）的简化实现。

该算法在节点数目非常多的场景下，效率远好于DistributedCache这种基于HDFS共享存储的方式。

下例中，两个文件一大一小，格式分别为：

Key, value, value
Key, value, value
1
2

Scala代码如下

 vartable1= sc.textFile(args(1))
 vartable2= sc.textFile(args(2))
 
 // table1 is smaller, so broadcast it as a map<String, String>
 varpairs =table1.map { x =>
   varpos =x.indexOf(',')
   (x.substring(0, pos), x.substring(pos + 1))
 }.collectAsMap
  
 varbroadCastMap =sc.broadcast(pairs) //save table1 as map, and broadcast it

 // table2 join table1 in map side
  
 varresult =table2.map { x =>
   varpos =x.indexOf(',')
   (x.substring(0, pos), x.substring(pos + 1))
 }.mapPartitions({ iter =>
   varm =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
  
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

Reduce-side Join

当两个文件非常大，难以将其中之一放到内存时，就可以采用Reduce-side Join。该算法需要通过Map和Reduce两个阶段完成，在Map阶段，将key相同的记录划分给同一个Reduce Task（需标记每条记录的来源，便于在Reduce阶段合并），在Reduce阶段，对key相同的进行合并。

Spark提供了Join算子，可以直接通过该算子实现reduce-side join，但要求RDD中的记录必须是key-value pairs，前面那个例子采用Reduce-side join实现如下：

  vartable1= sc.textFile(args(1))
 vartable2= sc.textFile(args(2))
 
 varpairs =table1.map{x=>
   varpos =x.indexOf(',')
   (x.substring(0, pos), x.substring(pos + 1))
 }
 
 varresult =table2.map{x=>
   varpos =x.indexOf(',')
   (x.substring(0, pos), x.substring(pos + 1))
 }.join(pairs)
 result.saveAsTextFile(args(3))
1
2
3
4
5
6
7
8
9
10
11
12
13