MapReduce算法 – 反转排序(Order Inversion)

本文译自 MapReduce Algorithms – Order Inversion

译者注:在刚开始翻译的时候,我将Order Inversion按照字面意思翻译成“反序”或者“倒序”,但是翻译完整篇文章之后,我感觉到,将Order Inversion翻译成反序模式是不恰当的,根据本文的内容,很显然,Inversion并非是将顺序倒排的意思,而是如同Spring的IOC一样,表明的是一种控制权的反转。Spring将对象的实例化责任从业务代码反转给了框架,而在本文的模式中,在mapreduce的sorting过程中,原来由框架负责的数据的排序以及shuffle规则被用户定制化了,控制权从框架反转到了user,实际上这种模式就是由用户控制sorting过程的意思

本文是一系列有关MapReduce算法的文章中的一篇,这些算法都在《Data-Intensive Text Processing with MapReduce》中提到过。这系列文章在本文之前已经发表的有 本地聚合, 本地聚合二 和 建立共生矩阵。在这篇文章里我们要讨论的是排序反转模式。这种模式利用MapReduce的排序(sorting)阶段,让一部分数据提前发送到reducer端以利于后续计算,如果你对MapReduce了解不多,我劝你读下去,因为我将展示给你如何使用排序(sorting)和partitioner来实现我们的目的,这将会大有益处。

尽管已经有许多MapReduce框架提供了高层次的抽象,例如Hive和Pig,理解底层是如何运行的仍然是有好处的。反序模式出现在《Data-Intensive Text Processing with MapReduce》这本书的第三章, 为了说明反序模式,我们要用共生矩阵模式中出现过的配对方法。建立共生矩阵的时候我们可以记录下词共同出现的次数,我门会对配对方法做一个小小的修改,mapper不止输出诸如(“foo”,”bar”) 这样的词对,还会额外输出(“foo”,”*”)这样的词对,对于每个词都依此法办理,这样可以很容易的得出左边的这个词的总共出现次数,用这个就可以计算出相对频率。这种方法会带来两个问题,首先我们需要想办法保证让 (“foo”,”*”) 成为reducer 的第一条记录,其次我们要保证左边的词相同的所有的词对都被同一个reducer所处理,我们先来看mapper代码再解决这两个问题。

Mapper Code

首先我们要对mapper做一些有别于配对方法的修改。在每次循环的最后,输出了某个词的所有的词对之后,输出一个特殊的词对(“word”,”*”), 计数就是这个词作为左边词的词对出现的次数。

  1. public class PairsRelativeOccurrenceMapper
  2. extends Mapper<LongWritable, Text, WordPair, IntWritable> {
  3. private WordPair wordPair = new WordPair();
  4. private IntWritable ONE = new IntWritable(1);
  5. private IntWritable totalCount = new IntWritable();
  6.  
  7. @Override
  8. protected void map(LongWritable key, Text value, Context context)
  9. throws IOException, InterruptedException {
  10. int neighbors = context.getConfiguration().getInt(‘neighbors’, 2);
  11. String[] tokens = value.toString().split(‘\\s+’);
  12. if (tokens.length &gt; 1) {
  13. for (int i = 0; i &lt; tokens.length; i++) {
  14. tokens[i] = tokens[i].replaceAll(‘\\W+’,);
  15.  
  16. if(tokens[i].equals()){
  17. continue;
  18. }
  19.  
  20. wordPair.setWord(tokens[i]);
  21.  
  22. int start = (i neighbors &lt; 0) ? 0 : i neighbors;
  23. int end = (i + neighbors &gt;= tokens.length) ? tokens.length 1 : i + neighbors;
  24. for (int j = start; j &lt;= end; j++) {
  25. if (j == i) continue;
  26. wordPair.setNeighbor(tokens[j].replaceAll(‘\\W’,));
  27. context.write(wordPair, ONE);
  28. }
  29. wordPair.setNeighbor(‘*’);
  30. totalCount.set(end start);
  31. context.write(wordPair, totalCount);
  32. }
  33. }
  34. }
  35. }

现在我们找到了统计特定词出现次数的办法,我们还需要想办法让这个特定的词对称为reduce处理的第一条记录以便计算相对频度。我们可以通过修改WordPair对象的compareTo方法在MapReduce 的sorting阶段来实现这个目的。

修改排序

修改WordPair类的compareTo方法,让发现 “*” 为右词的对象排到前列。

  1. @Override
  2. public int compareTo(WordPair other) {
  3. int returnVal = this.word.compareTo(other.getWord());
  4. if(returnVal != 0){
  5. return returnVal;
  6. }
  7. if(this.neighbor.toString().equals(‘*’)){
  8. return 1;
  9. }else if(other.getNeighbor().toString().equals(‘*’)){
  10. return 1;
  11. }
  12. return this.neighbor.compareTo(other.getNeighbor());
  13. }

通过修改compareTo方法,我们可以保证含有特殊字符的WordPair 都排在比较靠前的位置并会首先被reducer处理。这引出了第二个问题,我们怎样使具有相同左词的所有WordPai对象被发送到同一个reducer? 答案是定制一个partitioner。

定制 Partitioner

用key的hashcode对reducer数取模,就把key分配到了不同的reducer,这就是shuffle过程。但我们的WordPair 对象包含2个词,计算整个对象的hashcode是行不通的。我们需要写一个自己的Partitioner, 它在选择将输出发送到哪个reducer的时候只考虑左边的词。

  1. public class WordPairPartitioner extends Partitioner&lt;WordPair,IntWritable&gt; {
  2.  
  3. @Override
  4. public int getPartition(WordPair wordPair, IntWritable intWritable, int numPartitions) {
  5. return wordPair.getWord().hashCode() % numPartitions;
  6. }
  7. }

Reducer

写一个reducer来实现倒序模式很简单。引入一个计数变量以及一个表示当前词的“current”变量。reducer会检查作为输入key的WordPair 右边是不是特殊字符“*”。假如左边的词不等于“current”表示的词就重置计数变量,并且计算current表示的词的总次数。然后处理下一个WordPair对象,在同一个current范围内,计数之和与各个不同右词的计数结合就可以得到相对频率。继续这个过程直到发现另一个词(左词)然后再重新开始。

  1. public class PairsRelativeOccurrenceReducer extends
  2. Reducer<WordPair, IntWritable, WordPair, DoubleWritable> {
  3. private DoubleWritable totalCount = new DoubleWritable();
  4. private DoubleWritable relativeCount = new DoubleWritable();
  5. private Text currentWord = new Text(‘NOT_SET’);
  6. private Text flag = new Text(‘*’);
  7.  
  8. @Override
  9. protected void reduce(WordPair key, Iterable values, Context context)
  10. throws IOException, InterruptedException {
  11. if (key.getNeighbor().equals(flag)) {
  12. if (key.getWord().equals(currentWord)) {
  13. totalCount.set(totalCount.get() + getTotalCount(values));
  14. } else {
  15. currentWord.set(key.getWord());
  16. totalCount.set(0);
  17. totalCount.set(getTotalCount(values));
  18. }
  19. } else {
  20. int count = getTotalCount(values);
  21. relativeCount.set((double) count / totalCount.get());
  22. context.write(key, relativeCount);
  23. }
  24. }
  25. private int getTotalCount(Iterable values) {
  26. int count = 0;
  27. for (IntWritable value : values) {
  28. count += value.get();
  29. }
  30. return count;
  31. }
  32. }

通过控制sort阶段的逻辑和建立定制partitioner,我们可以把执行计算的reducer需要的数据在计算所需的数据到达之前发送到reducer,虽然这里没有展示,不过combiner在MapReduce中是经常会用到的。而且这个方法(使用combiner)也是mapper端合并模式的的一个非常好的实现。

现在我们可以保证有着相同左词的所有WordPair对象都被发到了同一个reducer。剩下的就是建立一个reducer来使用发送到reducer的数据。

例子和结果

在假期的这段时间里,我用查尔斯狄更斯的小说《圣诞颂歌》作为样例来运行了一下反序模式。我知道这可能没什么实际意义,但我们的目的就是这样。

  1. newhost2:sbin bbejeck$ hdfs dfs cat relative/part* | grep Humbug
  2. {word=[Humbug] neighbor=[Scrooge]} 0.2222222222222222
  3. {word=[Humbug] neighbor=[creation]} 0.1111111111111111
  4. {word=[Humbug] neighbor=[own]} 0.1111111111111111
  5. {word=[Humbug] neighbor=[said]} 0.2222222222222222
  6. {word=[Humbug] neighbor=[say]} 0.1111111111111111
  7. {word=[Humbug] neighbor=[to]} 0.1111111111111111
  8. {word=[Humbug] neighbor=[with]} 0.1111111111111111
  9. {word=[Scrooge] neighbor=[Humbug]} 0.0020833333333333333
  10. {word=[creation] neighbor=[Humbug]} 0.1
  11. {word=[own] neighbor=[Humbug]} 0.006097560975609756
  12. {word=[said] neighbor=[Humbug]} 0.0026246719160104987
  13. {word=[say] neighbor=[Humbug]} 0.010526315789473684
  14. {word=[to] neighbor=[Humbug]} 3.97456279809221E-4
  15. {word=[with] neighbor=[Humbug]} 9.372071227741331E-4

结论

即使在工作中计算相对词频的需求可能并不常见,我们也能够用这个来展示sorting和定制partitioner的用法,这可是我们写 MapReduce 程序时候的得力工具。如前所述,即使你的MapReduce都是用像Hive和Pig这样的高层次抽象语言写成的,了解一些底层的机制仍然是有好处的,谢谢。

    原文作者:排序算法
    原文地址: https://blog.csdn.net/hadoop_mapreduce/article/details/39802287
    本文转自网络文章,转载此文章仅为分享知识,如有侵权,请联系博主进行删除。
点赞