Hadoop MapReduce进阶 使用分布式缓存进行replicated join

轩辕阁

概念：
　　reduce-side join技术是灵活的，但是有时候它仍然会变得效率极低。由于join直到reduce()阶段才会开始，我们将会在网络中传递shuffle所有数据，而在大多数情况下，我们会在join阶段丢掉大多数传递的数据。因此我们期望能够在map阶段完成整个join操作。


主要技术难点：

在map阶段完成join的主要困难就是mapper可能需要与一个它自己不能获得的数据进行join操作，如果我们能够保证这样子的数据可被mapper获得，那我们这个技术就可用。举个例子，如果我们知道两个源数据被分为同样大小的partition，而且每个partition都以适合作为join
key的key值排序的话，那每个mapper()就可以获取所有join操作需要的数据。事实上，Hadoop的org.apache.hadoop.mared.join包中包含了这样的帮助类来实现mapside join，但不幸的是，这样的情况太少了。而且使用这样的类会造成额外的开销。因此，我们不会继续讨论这个包。





什么情况下使用？

情况1：如果我们知道两个源数据被分为同样大小的partition，而且每个partition都以适合作为join key的key值排序





情况2：当join大型数据时，通常只有一个源数据十分巨大，另一个数据可能就会呈数量级的减小。例如，一个电话公司的用户数据可能只有千万条用户数据，但他的交易记录数据可能会有十亿条数量级以上的具体电话记录。当小的数据源可以被分配到mapper的内存中时，我们可以获得效果明显的性能提高，只要通过将小的数据源拷贝到每一台mapper机器上，使mapper在map阶段就进行join操作。这个操作就叫做replicate
join。





解决方案：

Hadoop有一个叫做分布式缓存(distributed cache)的机制来将数据分发到集群上的所有节点上。它通常用来分发所有mapper需要的包含“background”数据的文件。例如你使用Hadoop来分类文档，你可能会有一个关键字的列表，你将使用distributed
cache来保证所有mapper能够获得这些keywords（"background data"）。


操作步骤：

1.将数据分发到每个节点上：

[java]
view plaincopyprint?

• DistributedCache.addCacheFile(newPath(args[0]).toUri(),conf);
  

DistributedCache.addCacheFile(new Path(args[0]).toUri(), conf);

2.在每个mapper上使用DistributedCache.getLocalCacheFiles()来获取文件，之后再进行相应的操作：


[java]
view plaincopyprint?

• DistributedCache.getLocalCacheFiles();
  

DistributedCache.getLocalCacheFiles();





新出现的问题：

我们的又一个限制是我们其中一个join的表必须足够小以至于能保存到内存中。尽管在不对称大小的输入数据中，较小的那个数据可能仍然不够小（不够小到可以放入内存中。）


1.我们可以通过重新安排数据加工步骤来使它们有效。例如：如果你需要所有用户在415区的排序数据时，在滤除一定记录前就将Orders以及Customers表连接起来虽然正确，但是效率却不高。Customers和Orders表都可能大到不能放入内存中。此时我们可以预处理数据使Customers或者Orders表变小。

2.有时候我们不论怎样预处理数据都不能使数据足够小，那我们应该在map时过滤掉不属于415区的用户。详见《Hadoop in Action》 Chapter5.2.3
semijoin