mapreduce 算法_MapReduce算法

mapreduce 算法

这篇文章是介绍MapReduce算法的系列文章的另一部分，该书在使用MapReduce进行数据密集型文本处理中找到。先前的文章是Local Aggregation ， Local Aggregation PartII和创建共现矩阵。这次我们将讨论阶数反转模式。顺序反转模式利用的MapReduce来计算所需要的提前将被操纵的数据的减速推送数据的排序阶段..你关闭此作为MapReduce的边缘状态之前，我强烈推荐您阅读作为我们将讨论如何利用排序的优势，并使用自定义分区程序进行覆盖，这两个实用程序都是有用的工具。

尽管许多MapReduce程序是用较高级别的抽象（即Hive或Pig）编写的，但了解较低级的情况仍然有帮助。顺序反转模式可在使用MapReduce进行数据密集型文本处理的第3章中找到。。为了说明顺序倒置模式，我们将从同现矩阵模式中使用Pairs方法。创建同现矩阵时，我们会跟踪单词一起出现的总次数。在较高的层次上，我们采用“成对”方法并稍加改动，除了使映射器发出诸如（“ foo”，“ bar”）之类的单词对外，我们还将发出（“ foo”，“ *”），并且将对每个单词对执行此操作，因此我们可以轻松得出最左单词出现频率的总计数，并使用该计数来计算我们的相对频率。这种方法提出了两个具体问题。首先，我们需要找到一种方法来确保单词对（“ foo”，“ *”）首先到达减速器。其次，我们需要确保所有具有相同左单词的单词对都到达相同的缩减词。在解决这些问题之前，让我们看一下我们的映射器代码。

映射器代码

首先，我们需要使用Pairs方法修改映射器。在发出特定单词的所有单词对之后，在每个循环的底部，我们将发出特殊标记WordPair（“ word”，“ *”）以及在左侧找到单词的次数。

public class PairsRelativeOccurrenceMapper extends Mapper<LongWritable, Text, WordPair, IntWritable> {private WordPair wordPair = new WordPair();private IntWritable ONE = new IntWritable(1);private IntWritable totalCount = new IntWritable();@Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {int neighbors = context.getConfiguration().getInt('neighbors', 2);String[] tokens = value.toString().split('\\s+');if (tokens.length > 1) {for (int i = 0; i < tokens.length; i++) {tokens[i] = tokens[i].replaceAll('\\W+','');if(tokens[i].equals('')){continue;}wordPair.setWord(tokens[i]);int start = (i - neighbors < 0) ? 0 : i - neighbors;int end = (i + neighbors >= tokens.length) ? tokens.length - 1 : i + neighbors;for (int j = start; j <= end; j++) {if (j == i) continue;wordPair.setNeighbor(tokens[j].replaceAll('\\W',''));context.write(wordPair, ONE);}wordPair.setNeighbor('*');totalCount.set(end - start);context.write(wordPair, totalCount);}}}
}

现在，我们已经生成了一种跟踪遇到一个特定单词的总次数的方法，我们需要确保那些特殊字符首先到达化简器，以便可以计算出总数以计算相对频率。通过修改WordPair对象上的compareTo方法，我们将在MapReduce流程的排序阶段为我们处理此问题。

修改后的排序

我们修改WordPair类上的compareTo方法，以便在右侧遇到“ *”字符时，将特定对象推到顶部。

@Overridepublic int compareTo(WordPair other) {int returnVal = this.word.compareTo(other.getWord());if(returnVal != 0){return returnVal;}if(this.neighbor.toString().equals('*')){return -1;}else if(other.getNeighbor().toString().equals('*')){return 1;}return this.neighbor.compareTo(other.getNeighbor());}

通过修改compareTo方法，我们现在可以确保将具有特殊字符的所有WordPair排在最前面，然后首先到达reducer。这导致了我们的第二个专业化，我们如何保证具有给定左单词的所有WordPair对象都将被发送到相同的reducer？答案是创建一个自定义分区程序。

自定义分区

通过计算键的哈希码对还原器的数量取模，将中间键改编为还原器。但是我们的WordPair对象包含两个单词，因此采用整个对象的哈希码显然是行不通的。我们需要修改一个自定义的分区程序，该分区程序在确定将哪个缩减程序发送到输出时仅考虑左边的单词。

public class WordPairPartitioner extends Partitioner<WordPair,IntWritable> {@Overridepublic int getPartition(WordPair wordPair, IntWritable intWritable, int numPartitions) {return wordPair.getWord().hashCode() % numPartitions;}
}

现在，我们保证将所有具有相同左单词的WordPair对象发送到相同的reducer。剩下的就是构造一个化简器以利用发送数据的格式。

减速器

为倒序反转模式构建减速器很简单。这将涉及保持计数器变量和“当前”字变量。减速器将检查输入键WordPair中右侧的特殊字符“ *”。如果左边的单词不等于“当前”单词，我们将重新设置计数器并求和所有值，以获得观察到给定当前单词的总次数。现在，我们将处理下一个WordPair对象，对计数求和并除以我们的计数器变量，以获得相对频率。该过程将继续进行，直到遇到另一个特殊字符并重新开始。

public class PairsRelativeOccurrenceReducer extends Reducer<WordPair, IntWritable, WordPair, DoubleWritable> {private DoubleWritable totalCount = new DoubleWritable();private DoubleWritable relativeCount = new DoubleWritable();private Text currentWord = new Text('NOT_SET');private Text flag = new Text('*');@Overrideprotected void reduce(WordPair key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {if (key.getNeighbor().equals(flag)) {if (key.getWord().equals(currentWord)) {totalCount.set(totalCount.get() + getTotalCount(values));} else {currentWord.set(key.getWord());totalCount.set(0);totalCount.set(getTotalCount(values));}} else {int count = getTotalCount(values);relativeCount.set((double) count / totalCount.get());context.write(key, relativeCount);}}private int getTotalCount(Iterable<IntWritable> values) {int count = 0;for (IntWritable value : values) {count += value.get();}return count;}
}

通过处理排序顺序并创建自定义分区程序，我们已经能够在计算所需的数据到达之前将数据发送到计算所需的化简器。尽管此处未显示，但使用组合器来运行MapReduce作业。这种方法也是“映射器内”合并模式的不错选择。

示例与结果

鉴于假期即将来临，我觉得现在是时候针对查尔斯·狄更斯（Charles Dickens）的小说《圣诞节颂歌》（A Christmas Carol）进行订单倒置模式的例子了。我知道这很老套，但它能达到目的。

new-host-2:sbin bbejeck$ hdfs dfs -cat relative/part* | grep Humbug
{word=[Humbug] neighbor=[Scrooge]}	0.2222222222222222
{word=[Humbug] neighbor=[creation]}	0.1111111111111111
{word=[Humbug] neighbor=[own]}	0.1111111111111111
{word=[Humbug] neighbor=[said]}	0.2222222222222222
{word=[Humbug] neighbor=[say]}	0.1111111111111111
{word=[Humbug] neighbor=[to]}	0.1111111111111111
{word=[Humbug] neighbor=[with]}	0.1111111111111111
{word=[Scrooge] neighbor=[Humbug]}	0.0020833333333333333
{word=[creation] neighbor=[Humbug]}	0.1
{word=[own] neighbor=[Humbug]}	0.006097560975609756
{word=[said] neighbor=[Humbug]}	0.0026246719160104987
{word=[say] neighbor=[Humbug]}	0.010526315789473684
{word=[to] neighbor=[Humbug]}	3.97456279809221E-4
{word=[with] neighbor=[Humbug]}	9.372071227741331E-4