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

  • 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的算法如下:


var table1 = sc.textFile(args(1))
    var table2 = sc.textFile(args(2))
    
    // table1 is smaller, so broadcast it as a map<String, String>
    var pairs = table1.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 = table2.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 or HDFS
  • 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实现如下:

var table1 = sc.textFile(args(1))
    var table2 = sc.textFile(args(2))
    
    var pairs = table1.map{x =>
      var pos = x.indexOf(',')
      (x.substring(0, pos), x.substring(pos + 1))
    }

    var result = table2.map{x =>
      var pos = x.indexOf(',')
      (x.substring(0, pos), x.substring(pos + 1))
    }.join(pairs)

    result.saveAsTextFile(args(3))
  • 总结

本文介绍了Spark中map-side join和reduce-side join的编程思路,希望对大家有借鉴意义。但需要注意的是,在使用这两种算法处理较大规模的数据时,通常需要对多个参数进行调优,否则可能会产生OOM问题。通常需要调优的相关参数包括,map端数据输出buffer大小,reduce端数据分组方法(基于map还是基于sort),等等。

原创文章,转载请注明: 转载自董的博客

本文链接地址: Apache Spark探秘:实现Map-side Join和Reduce-side Join

微信公众号:hadoop-123,专注于大数据技术分享,欢迎加入!

说点什么

avatar
  Subscribe  
提醒