mapreduce分类算法

⑴ maprece 实现bayes或Kmeans,lD3分类算法

这个自己写多累啊 建议你去看一下mahout 他们实现了很多的经典机器学习算法，用MpaRece编程模型，都是可以运行在Hadoop集群上的

⑵ maprece 可以做哪些机器学习算法

maprece其实不适合做机器学习，更适合进行大规模数据的处理
因为机器学习是计算密集型的任务，通常需要反复的迭代，而maprece中间数据存放在磁盘上，速度很慢。
机器学习算法建议使用MPI框架或者Spark ML

⑶ 如何实现maprece计算框架以有效实现迭代

MapRece从出现以来，已经成为Apache Hadoop计算范式的扛鼎之作。它对于符合其设计的各项工作堪称完美：大规模日志处理，ETL批处理操作等。

随着Hadoop使用范围的不断扩大，人们已经清楚知道MapRece不是所有计算的最佳框架。Hadoop 2将资源管理器YARN作为自己的顶级组件，为其他计算引擎的接入提供了可能性。如Impala等非MapRece架构的引入，使平台具备了支持交互式SQL的能力。

今天，Apache Spark是另一种这样的替代，并且被称为是超越MapRece的通用计算范例。也许您会好奇：MapRece一直以来已经这么有用了，怎么能突然被取代？毕竟，还有很多ETL这样的工作需要在Hadoop上进行，即使该平台目前也已经拥有其他实时功能。

值得庆幸的是，在Spark上重新实现MapRece一样的计算是完全可能的。它们可以被更简单的维护，而且在某些情况下更快速，这要归功于Spark优化了刷写数据到磁盘的过程。Spark重新实现MapRece编程范式不过是回归本源。Spark模仿了Scala的函数式编程风格和API。而MapRece的想法来自于函数式编程语言LISP。

尽管Spark的主要抽象是RDD（弹性分布式数据集），实现了Map，rece等操作，但这些都不是Hadoop的Mapper或Recer API的直接模拟。这些转变也往往成为开发者从Mapper和Recer类平行迁移到Spark的绊脚石。

与Scala或Spark中经典函数语言实现的map和rece函数相比，原有Hadoop提供的Mapper和Recer API 更灵活也更复杂。这些区别对于习惯了MapRece的开发者而言也许并不明显，下列行为是针对Hadoop的实现而不是MapRece的抽象概念：
· Mapper和Recer总是使用键值对作为输入输出。
· 每个Recer按照Key对Value进行rece。
· 每个Mapper和Recer对于每组输入可能产生0个，1个或多个键值对。
· Mapper和Recer可能产生任意的keys和values，而不局限于输入的子集和变换。
Mapper和Recer对象的生命周期可能横跨多个map和rece操作。它们支持setup和cleanup方法，在批量记录处理开始之前和结束之后被调用。

本文将简要展示怎样在Spark中重现以上过程，您将发现不需要逐字翻译Mapper和Recer！

作为元组的键值对
假定我们需要计算大文本中每一行的长度，并且报告每个长度的行数。在HadoopMapRece中，我们首先使用一个Mapper，生成为以行的长度作为key，1作为value的键值对。
public class LineLengthMapper extends
Mapper<LongWritable, Text, IntWritable, IntWritable> {
@Override
protected void map(LongWritable lineNumber, Text line, Context context)
throws IOException, InterruptedException {
context.write(new IntWritable(line.getLength()), new IntWritable(1));
}
}

值得注意的是Mappers和Recers只对键值对进行操作。所以由TextInputFormat提供输入给LineLengthMapper，实际上也是以文本中位置为key（很少这么用，但是总是需要有东西作为Key），文本行为值的键值对。

与之对应的Spark实现：

lines.map(line => (line.length, 1))

Spark中，输入只是String构成的RDD，而不是key-value键值对。Spark中对key-value键值对的表示是一个Scala的元组，用(A，B)这样的语法来创建。上面的map操作的结果是(Int，Int)元组的RDD。当一个RDD包含很多元组，它获得了多个方法，如receByKey，这对再现MapRece行为将是至关重要的。

Rece
rece()与receBykey()
统计行的长度的键值对，需要在Recer中对每种长度作为key，计算其行数的总和作为value。
public class LineLengthRecer extends
Recer<IntWritable, IntWritable, IntWritable, IntWritable> {
@Override
protected void rece(IntWritable length, Iterable<IntWritable> counts,
Context context) throws IOException, InterruptedException {
int sum = 0;
for (IntWritable count : counts) {
sum += count.get();
}
context.write(length, new IntWritable(sum));
}
}

Spark中与上述Mapper，Recer对应的实现只要一行代码：
val lengthCounts = lines.map(line => (line.length, 1)).receByKey(_ + _)

Spark的RDD API有个rece方法，但是它会将所有key-value键值对rece为单个value。这并不是Hadoop MapRece的行为，Spark中与之对应的是ReceByKey。

另外，Recer的Rece方法接收多值流，并产生0，1或多个结果。而receByKey，它接受的是一个将两个值转化为一个值的函数，在这里，就是把两个数字映射到它们的和的简单加法函数。此关联函数可以被调用者用来rece多个值到一个值。与Recer方法相比，他是一个根据Key来Rece Value的更简单而更精确的API。
Mapper
map() 与 flatMap()
现在，考虑一个统计以大写字母开头的单词的个数的算法。对于每行输入文本，Mapper可能产生0个，1个或多个键值对。
public class CountUppercaseMapper extends
Mapper<LongWritable, Text, Text, IntWritable> {
@Override
protected void map(LongWritable lineNumber, Text line, Context context)
throws IOException, InterruptedException {
for (String word : line.toString().split(" ")) {
if (Character.isUpperCase(word.charAt(0))) {
context.write(new Text(word), new IntWritable(1));
}
}
}
}

Spark对应的写法：
lines.flatMap(
_.split(" ").filter(word => Character.isUpperCase(word(0))).map(word => (word,1))
)
简单的Spark map函数不适用于这种场景，因为map对于每个输入只能产生单个输出，但这个例子中一行需要产生多个输出。所以，和MapperAPI支持的相比，Spark的map函数语义更简单，应用范围更窄。

Spark的解决方案是首先将每行映射为一组输出值，这组值可能为空值或多值。随后会通过flatMap函数被扁平化。数组中的词会被过滤并被转化为函数中的元组。这个例子中，真正模仿Mapper行为的是flatMap，而不是map。
groupByKey()
写一个统计次数的recer是简单的，在Spark中，receByKey可以被用来统计每个单词的总数。比如出于某种原因要求输出文件中每个单词都要显示为大写字母和其数量，在MapRece中，实现如下：
public class CountUppercaseRecer extends
Recer<Text, IntWritable, Text, IntWritable> {
@Override
protected void rece(Text word, Iterable<IntWritable> counts, Context context)
throws IOException, InterruptedException {
int sum = 0;
for (IntWritable count : counts) {
sum += count.get();
}
context
.write(new Text(word.toString().toUpperCase()), new IntWritable(sum));
}
}

但是redeceByKey不能单独在Spark中工作，因为他保留了原来的key。为了在Spark中模拟，我们需要一些更像Recer API的操作。我们知道Recer的rece方法接受一个key和一组值，然后完成一组转换。groupByKey和一个连续的map操作能够达到这样的目标:
groupByKey().map { case (word,ones) => (word.toUpperCase, ones.sum) }
groupByKey只是将某一个key的所有值收集在一起，并且不提供rece功能。以此为基础，任何转换都可以作用在key和一系列值上。此处，将key转变为大写字母，将values直接求和。

setup()和cleanup()

在MapRece中，Mapper和Recer可以声明一个setup方法，在处理输入之前执行，来进行分配数据库连接等昂贵资源，同时可以用cleanup函数可以释放资源。
public class SetupCleanupMapper extends
Mapper<LongWritable, Text, Text, IntWritable> {
private Connection dbConnection;
@Override
protected void setup(Context context) {
dbConnection = ...;
}
...
@Override
protected void cleanup(Context context) {
dbConnection.close();
}
}

Spark中的map和flatMap方法每次只能在一个input上操作，而且没有提供在转换大批值前后执行代码的方法，看起来，似乎可以直接将setup和cleanup代码放在Sparkmap函数调用之前和之后：
val dbConnection = ...
lines.map(... dbConnection.createStatement(...) ...)
dbConnection.close() // Wrong!

然而这种方法却不可行，原因在于：
· 它将对象dbConnection放在map函数的闭包中，这需要他是可序列化的（比如，通过java.io.Serializable实现）。而数据库连接这种对象一般不能被序列化。
· map是一种转换，而不是操作，并且拖延执行。连接对象不能被及时关闭。
· 即便如此，它也只能关闭driver上的连接，而不是释放被序列化拷贝版本分配的资源连接。

事实上，map和flatMap都不是Spark中Mapper的最接近的对应函数，Spark中Mapper的最接近的对应函数是十分重要的mapPartitions()方法，这个方法能够不仅完成单值对单值的映射，也能完成一组值对另一组值的映射，很像一个批映射（bulkmap）方法。这意味着mapPartitions()方法能够在开始时从本地分配资源，并在批映射结束时释放资源。

添加setup方法是简单的，添加cleanup会更困难，这是由于检测转换完成仍然是困难的。例如，这样是能工作的：

lines.mapPartitions { valueIterator =>
val dbConnection = ... // OK
val transformedIterator = valueIterator.map(... dbConnection ...)
dbConnection.close() // Still wrong! May not have evaluated iterator
transformedIterator
}

一个完整的范式应该看起来类似于：
lines.mapPartitions { valueIterator =>
if (valueIterator.isEmpty) {
Iterator[...]()
} else {
val dbConnection = ...
valueIterator.map { item =>
val transformedItem = ...
if (!valueIterator.hasNext) {
dbConnection.close()
}
transformedItem
}
}
}

虽然后者代码翻译注定不如前者优雅，但它确实能够完成工作。

flatMapPartitions方法并不存在，然而，可以通过调用mapPartitions，后面跟一个flatMap(a= > a)的调用达到同样效果。

带有setup和cleanup的Recer对应只需仿照上述代码使用groupByKey后面跟一个mapPartition函数。

别急，等一下，还有更多
MapRece的开发者会指出，还有更多的还没有被提及的API：
· MapRece支持一种特殊类型的Recer，也称为Combiner，可以从Mapper中减少洗牌(shuffled)数据大小。
· 它还支持同通过Partitioner实现的自定义分区，和通过分组Comparator实现的自定义分组。
· Context对象授予Counter API的访问权限以及它的累积统计。
· Recer在其生命周期内一直能看到已排序好的key 。
· MapRece有自己的Writable序列化方案。
· Mapper和Recer可以一次发射多组输出。
· MapRece有几十个调优参数。

有很多方法可以在Spark中实现这些方案，使用类似Accumulator的API，类似groupBy和在不同的这些方法中加入partitioner参数的方法，Java或Kryo序列化，缓存和更多。由于篇幅限制，在这篇文章中就不再累赘介绍了。

需要指出的是，MapRece的概念仍然有用。只不过现在有了一个更强大的实现，并利用函数式语言，更好地匹配其功能性。理解Spark RDD API和原来的Mapper和RecerAPI之间的差异，可以帮助开发者更好地理解所有这些函数的工作原理，以及理解如何利用Spark发挥其优势。

⑷ Hadoop和MapRece究竟分别是做什么用的

Hadoop是用来开发分布式程序的架构，是一个由Apache基金会所开发的分布式系统基础架构。用户可以在不了解分布式底层细节的情况下，开发分布式程序。

MapRece是用来做大规模并行数据处理的数据模型。方便了编程人员在不会分布式并行编程的情况下，将自己的程序运行在分布式系统上。

(4)maprece分类算法扩展阅读

Hadoop是一个能够让用户轻松架构和使用的分布式计算平台。用户可以轻松地在Hadoop上开发和运行处理海量数据的应用程序。主要有以下几个优点 ：

1、高可靠性。Hadoop按位存储和处理数据的能力值得人们信赖 。

2、高扩展性。Hadoop是在可用的计算机集簇间分配数据并完成计算任务的，这些集簇可以方便地扩展到数以千计的节点中 。

3、高效性。Hadoop能够在节点之间动态地移动数据，并保证各个节点的动态平衡，因此处理速度非常快 。

4、高容错性。Hadoop能够自动保存数据的多个副本，并且能够自动将失败的任务重新分配。

5、低成本。与一体机、商用数据仓库以及QlikView、Yonghong Z-Suite等数据集市相比，hadoop是开源的，项目的软件成本因此会大大降低。

⑸ 什么是Map/Rece-Maprece-about云开发

什么是Map/Rece，看下面的各种解释：

(1)MapRece是hadoop的核心组件之一，hadoop要分布式包括两部分，一是分布式文件系统hdfs,一部是分布式计算框，就是maprece,缺一不可，也就是说，可以通过maprece很容易在hadoop平台上进行分布式的计算编程。

(2)Maprece是一种编程模型，是一种编程方法，抽象理论。

(3)下面是一个关于一个程序员是如何个妻子讲解什么是MapRece？文章很长请耐心的看。

我问妻子：“你真的想要弄懂什么是MapRece？” 她很坚定的回答说“是的”。 因此我问道：

我： 你是如何准备洋葱辣椒酱的？（以下并非准确食谱，请勿在家尝试）

妻子： 我会取一个洋葱，把它切碎，然后拌入盐和水，最后放进混合研磨机里研磨。这样就能得到洋葱辣椒酱了。

妻子： 但这和MapRece有什么关系？

我： 你等一下。让我来编一个完整的情节，这样你肯定可以在15分钟内弄懂MapRece.

妻子： 好吧。

我：现在，假设你想用薄荷、洋葱、番茄、辣椒、大蒜弄一瓶混合辣椒酱。你会怎么做呢？

妻子： 我会取薄荷叶一撮，洋葱一个，番茄一个，辣椒一根，大蒜一根，切碎后加入适量的盐和水，再放入混合研磨机里研磨，这样你就可以得到一瓶混合辣椒酱了。

我： 没错，让我们把MapRece的概念应用到食谱上。Map和Rece其实是两种操作，我来给你详细讲解下。
Map（映射）: 把洋葱、番茄、辣椒和大蒜切碎，是各自作用在这些物体上的一个Map操作。所以你给Map一个洋葱，Map就会把洋葱切碎。 同样的，你把辣椒，大蒜和番茄一一地拿给Map，你也会得到各种碎块。 所以，当你在切像洋葱这样的蔬菜时，你执行就是一个Map操作。 Map操作适用于每一种蔬菜，它会相应地生产出一种或多种碎块，在我们的例子中生产的是蔬菜块。在Map操作中可能会出现有个洋葱坏掉了的情况，你只要把坏洋葱丢了就行了。所以，如果出现坏洋葱了，Map操作就会过滤掉坏洋葱而不会生产出任何的坏洋葱块。

Rece（化简）:在这一阶段，你将各种蔬菜碎都放入研磨机里进行研磨，你就可以得到一瓶辣椒酱了。这意味要制成一瓶辣椒酱，你得研磨所有的原料。因此，研磨机通常将map操作的蔬菜碎聚集在了一起。

妻子： 所以，这就是MapRece?

我： 你可以说是，也可以说不是。 其实这只是MapRece的一部分，MapRece的强大在于分布式计算。

妻子： 分布式计算？ 那是什么？请给我解释下吧。

我： 没问题。

我： 假设你参加了一个辣椒酱比赛并且你的食谱赢得了最佳辣椒酱奖。得奖之后，辣椒酱食谱大受欢迎，于是你想要开始出售自制品牌的辣椒酱。假设你每天需要生产10000瓶辣椒酱，你会怎么办呢？

妻子： 我会找一个能为我大量提供原料的供应商。

我：是的..就是那样的。那你能否独自完成制作呢？也就是说，独自将原料都切碎？ 仅仅一部研磨机又是否能满足需要？而且现在，我们还需要供应不同种类的辣椒酱，像洋葱辣椒酱、青椒辣椒酱、番茄辣椒酱等等。

妻子： 当然不能了，我会雇佣更多的工人来切蔬菜。我还需要更多的研磨机，这样我就可以更快地生产辣椒酱了。
我：没错，所以现在你就不得不分配工作了，你将需要几个人一起切蔬菜。每个人都要处理满满一袋的蔬菜，而每一个人都相当于在执行一个简单的Map操作。每一个人都将不断的从袋子里拿出蔬菜来，并且每次只对一种蔬菜进行处理，也就是将它们切碎，直到袋子空了为止。
这样，当所有的工人都切完以后，工作台（每个人工作的地方）上就有了洋葱块、番茄块、和蒜蓉等等。

妻子：但是我怎么会制造出不同种类的番茄酱呢？

我：现在你会看到MapRece遗漏的阶段—搅拌阶段。MapRece将所有输出的蔬菜碎都搅拌在了一起，这些蔬菜碎都是在以key为基础的 map操作下产生的。搅拌将自动完成，你可以假设key是一种原料的名字，就像洋葱一样。 所以全部的洋葱keys都会搅拌在一起，并转移到研磨洋葱的研磨器里。这样，你就能得到洋葱辣椒酱了。同样地，所有的番茄也会被转移到标记着番茄的研磨器里，并制造出番茄辣椒酱。

（4）上面都是从理论上来说明什么是MapRece，那么咱们在MapRece产生的过程和代码的角度来理解这个问题。
如果想统计下过去10年计算机论文出现最多的几个单词，看看大家都在研究些什么，那收集好论文后，该怎么办呢？

方法一：
我可以写一个小程序，把所有论文按顺序遍历一遍，统计每一个遇到的单词的出现次数，最后就可以知道哪几个单词最热门了。 这种方法在数据集比较小时，是非常有效的，而且实现最简单，用来解决这个问题很合适。

方法二：
写一个多线程程序，并发遍历论文。
这个问题理论上是可以高度并发的，因为统计一个文件时不会影响统计另一个文件。当我们的机器是多核或者多处理器，方法二肯定比方法一高效。但是写一个多线程程序要比方法一困难多了，我们必须自己同步共享数据，比如要防止两个线程重复统计文件。

方法三：
把作业交给多个计算机去完成。
我们可以使用方法一的程序，部署到N台机器上去，然后把论文集分成N份，一台机器跑一个作业。这个方法跑得足够快，但是部署起来很麻烦，我们要人工把程序到别的机器，要人工把论文集分开，最痛苦的是还要把N个运行结果进行整合（当然我们也可以再写一个程序）。

⑹ 请简要描述Hadoop计算框架MapRece的工作原理

分为2个步骤，map和rece，map专门负责对每个数据独立地同时地打标签，框架会对相同标签的数据分成一组，rece对分好的那些组数据做累计计算。我们只要分别实现map和rece就可以了

⑺ 如何使用MapRece计算相似度

et expansion），主要有如下几种方法（以Document Similarity为例）：Brute Force：最直接、暴力的方法，两个for循环，计算任意两篇文档之间的相似度，时间复杂度为O(n^2)。这种方法可以得到最好的效果，但是计算量太大，效率较差，往往作为baseline。
Inverted Index Based：由于大量文档之间没有交集term，为了优化算法性能，只需计算那些包含相同term文档之间的相似度即可，算法伪代码如下：基于MapRece的分布式计算框架如下：为了进一步优化计算，节省空间，研究人员提出了一系列剪枝策略和近似算法，详细见：《Scaling Up All Pairs Similarity Search》、《Pairwise document similarity in large collections with MapRece》、《Brute Force and Indexed Approaches to Pairwise Document Similarity Comparisons with MapRece》。
Locality Sensitive Hashing（LSH）：通过对文档进行某种度量操作后将其分组散列在不同的桶中。在这种度量下相似度较高的文档被分在同一个桶中的可能性较高。主要用于Near-plicate detection和Image similarity identification等，详细见：《Approximate Nearest Neighbors: Towards Removing the Curse of Dimensionality》、《Google news personalization: scalable online collaborative filtering》。

⑻ maprece运用什么排序算法

但是该方法在处理大型文件时效率极低，因为一台机器必须处理所有输出文件，从而完全丧失了MapRece所提供的并行架构的优势。

⑼ hadoop的maprece常见算法案例有几种

基本MapRece模式

计数与求和
问题陈述:
有许多文档，每个文档都有一些字段组成。需要计算出每个字段在所有文档中的出现次数或者这些字段的其他什么统计值。例如，给定一个log文件，其中的每条记录都包含一个响应时间，需要计算出平均响应时间。
解决方案:
让我们先从简单的例子入手。在下面的代码片段里，Mapper每遇到指定词就把频次记1，Recer一个个遍历这些词的集合然后把他们的频次加和。

1 class Mapper
2 method Map(docid id, doc d)
3 for all term t in doc d do
4 Emit(term t, count 1)
5
6 class Recer
7 method Rece(term t, counts [c1, c2,...])
8 sum = 0
9 for all count c in [c1, c2,...] do
10 sum = sum + c
11 Emit(term t, count sum)

这种方法的缺点显而易见，Mapper提交了太多无意义的计数。它完全可以通过先对每个文档中的词进行计数从而减少传递给Recer的数据量:

1 class Mapper
2 method Map(docid id, doc d)
3 H = new AssociativeArray
4 for all term t in doc d do
5 H{t} = H{t} + 1
6 for all term t in H do
7 Emit(term t, count H{t})

如果要累计计数的的不只是单个文档中的内容，还包括了一个Mapper节点处理的所有文档，那就要用到Combiner了:

1 class Mapper
2 method Map(docid id, doc d)
3 for all term t in doc d do
4 Emit(term t, count 1)
5
6 class Combiner
7 method Combine(term t, [c1, c2,...])
8 sum = 0
9 for all count c in [c1, c2,...] do
10 sum = sum + c
11 Emit(term t, count sum)
12
13 class Recer
14 method Rece(term t, counts [c1, c2,...])
15 sum = 0
16 for all count c in [c1, c2,...] do
17 sum = sum + c
18 Emit(term t, count sum)

应用：Log 分析, 数据查询

整理归类

问题陈述:
有一系列条目，每个条目都有几个属性，要把具有同一属性值的条目都保存在一个文件里，或者把条目按照属性值分组。 最典型的应用是倒排索引。
解决方案：
解决方案很简单。 在 Mapper 中以每个条目的所需属性值作为 key，其本身作为值传递给 Recer。 Recer 取得按照属性值分组的条目，然后可以处理或者保存。如果是在构建倒排索引，那么 每个条目相当于一个词而属性值就是词所在的文档ID。
应用：倒排索引， ETL
过滤 (文本查找)，解析和校验
问题陈述:
假设有很多条记录，需要从其中找出满足某个条件的所有记录，或者将每条记录传换成另外一种形式（转换操作相对于各条记录独立，即对一条记录的操作与其他记录无关）。像文本解析、特定值抽取、格式转换等都属于后一种用例。
解决方案:
非常简单，在Mapper 里逐条进行操作，输出需要的值或转换后的形式。
应用：日志分析，数据查询，ETL，数据校验

分布式任务执行

问题陈述:
大型计算可以分解为多个部分分别进行然后合并各个计算的结果以获得最终结果。
解决方案: 将数据切分成多份作为每个 Mapper 的输入，每个Mapper处理一份数据，执行同样的运算，产生结果，Recer把多个Mapper的结果组合成一个。
案例研究： 数字通信系统模拟
像 WiMAX 这样的数字通信模拟软件通过系统模型来传输大量的随机数据，然后计算传输中的错误几率。 每个 Mapper 处理样本 1/N 的数据，计算出这部分数据的错误率，然后在 Recer 里计算平均错误率。
应用：工程模拟，数字分析，性能测试
排序
问题陈述:
有许多条记录，需要按照某种规则将所有记录排序或是按照顺序来处理记录。
解决方案: 简单排序很好办 – Mappers 将待排序的属性值为键，整条记录为值输出。 不过实际应用中的排序要更加巧妙一点， 这就是它之所以被称为MapRece 核心的原因（“核心”是说排序？因为证明Hadoop计算能力的实验是大数据排序？还是说Hadoop的处理过程中对key排序的环节？）。在实践中，常用组合键来实现二次排序和分组。
MapRece 最初只能够对键排序， 但是也有技术利用可以利用Hadoop 的特性来实现按值排序。想了解的话可以看这篇博客。
按照BigTable的概念，使用 MapRece来对最初数据而非中间数据排序，也即保持数据的有序状态更有好处，必须注意这一点。换句话说，在数据插入时排序一次要比在每次查询数据的时候排序更高效。
应用：ETL，数据分析

非基本 MapRece 模式

迭代消息传递 (图处理)

问题陈述：
假设一个实体网络，实体之间存在着关系。 需要按照与它比邻的其他实体的属性计算出一个状态。这个状态可以表现为它和其它节点之间的距离， 存在特定属性的邻接点的迹象， 邻域密度特征等等。
解决方案：
网络存储为系列节点的结合，每个节点包含有其所有邻接点ID的列表。按照这个概念，MapRece 迭代进行，每次迭代中每个节点都发消息给它的邻接点。邻接点根据接收到的信息更新自己的状态。当满足了某些条件的时候迭代停止，如达到了最大迭代次数（网络半径）或两次连续的迭代几乎没有状态改变。从技术上来看，Mapper 以每个邻接点的ID为键发出信息，所有的信息都会按照接受节点分组，recer 就能够重算各节点的状态然后更新那些状态改变了的节点。下面展示了这个算法：

1 class Mapper
2 method Map(id n, object N)
3 Emit(id n, object N)
4 for all id m in N.OutgoingRelations do
5 Emit(id m, message getMessage(N))
6
7 class Recer
8 method Rece(id m, [s1, s2,...])
9 M = null
10 messages = []
11 for all s in [s1, s2,...] do
12 if IsObject(s) then
13 M = s
14 else // s is a message
15 messages.add(s)
16 M.State = calculateState(messages)
17 Emit(id m, item M)

一个节点的状态可以迅速的沿着网络传全网，那些被感染了的节点又去感染它们的邻居，整个过程就像下面的图示一样：

案例研究： 沿分类树的有效性传递
问题陈述：
这个问题来自于真实的电子商务应用。将各种货物分类，这些类别可以组成一个树形结构，比较大的分类（像男人、女人、儿童）可以再分出小分类（像男裤或女装），直到不能再分为止（像男式蓝色牛仔裤）。这些不能再分的基层类别可以是有效（这个类别包含有货品）或者已无效的（没有属于这个分类的货品）。如果一个分类至少含有一个有效的子分类那么认为这个分类也是有效的。我们需要在已知一些基层分类有效的情况下找出分类树上所有有效的分类。
解决方案：
这个问题可以用上一节提到的框架来解决。我们咋下面定义了名为 getMessage和 calculateState 的方法：

1 class N
2 State in {True = 2, False = 1, null = 0},
3 initialized 1 or 2 for end-of-line categories, 0 otherwise
4 method getMessage(object N)
5 return N.State
6 method calculateState(state s, data [d1, d2,...])
7 return max( [d1, d2,...] )

案例研究：广度优先搜索
问题陈述：需要计算出一个图结构中某一个节点到其它所有节点的距离。
解决方案： Source源节点给所有邻接点发出值为0的信号，邻接点把收到的信号再转发给自己的邻接点，每转发一次就对信号值加1：

1 class N
2 State is distance,
3 initialized 0 for source node, INFINITY for all other nodes
4 method getMessage(N)
5 return N.State + 1
6 method calculateState(state s, data [d1, d2,...])
7 min( [d1, d2,...] )

案例研究：网页排名和 Mapper 端数据聚合
这个算法由Google提出，使用权威的PageRank算法，通过连接到一个网页的其他网页来计算网页的相关性。真实算法是相当复杂的，但是核心思想是权重可以传播，也即通过一个节点的各联接节点的权重的均值来计算节点自身的权重。

1 class N
2 State is PageRank
3 method getMessage(object N)
4 return N.State / N.OutgoingRelations.size()
5 method calculateState(state s, data [d1, d2,...])
6 return ( sum([d1, d2,...]) )

要指出的是上面用一个数值来作为评分实际上是一种简化，在实际情况下，我们需要在Mapper端来进行聚合计算得出这个值。下面的代码片段展示了这个改变后的逻辑 （针对于 PageRank 算法）：

1 class Mapper
2 method Initialize
3 H = new AssociativeArray
4 method Map(id n, object N)
5 p = N.PageRank / N.OutgoingRelations.size()
6 Emit(id n, object N)
7 for all id m in N.OutgoingRelations do
8 H{m} = H{m} + p
9 method Close
10 for all id n in H do
11 Emit(id n, value H{n})
12
13 class Recer
14 method Rece(id m, [s1, s2,...])
15 M = null
16 p = 0
17 for all s in [s1, s2,...] do
18 if IsObject(s) then
19 M = s
20 else
21 p = p + s
22 M.PageRank = p
23 Emit(id m, item M)

应用：图分析，网页索引

值去重 （对唯一项计数）
问题陈述: 记录包含值域F和值域 G，要分别统计相同G值的记录中不同的F值的数目 (相当于按照 G分组).
这个问题可以推而广之应用于分面搜索（某些电子商务网站称之为Narrow Search）
Record 1: F=1, G={a, b}
Record 2: F=2, G={a, d, e}
Record 3: F=1, G={b}
Record 4: F=3, G={a, b}

Result:
a -> 3 // F=1, F=2, F=3
b -> 2 // F=1, F=3
d -> 1 // F=2
e -> 1 // F=2

解决方案 I:
第一种方法是分两个阶段来解决这个问题。第一阶段在Mapper中使用F和G组成一个复合值对，然后在Recer中输出每个值对，目的是为了保证F值的唯一性。在第二阶段，再将值对按照G值来分组计算每组中的条目数。
第一阶段：

1 class Mapper
2 method Map(null, record [value f, categories [g1, g2,...]])
3 for all category g in [g1, g2,...]
4 Emit(record [g, f], count 1)
5
6 class Recer
7 method Rece(record [g, f], counts [n1, n2, ...])
8 Emit(record [g, f], null )

第二阶段：

1 class Mapper
2 method Map(record [f, g], null)
3 Emit(value g, count 1)
4
5 class Recer
6 method Rece(value g, counts [n1, n2,...])
7 Emit(value g, sum( [n1, n2,...] ) )

解决方案 II:
第二种方法只需要一次MapRece 即可实现，但扩展性不强。算法很简单-Mapper 输出值和分类，在Recer里为每个值对应的分类去重然后给每个所属的分类计数加1，最后再在Recer结束后将所有计数加和。这种方法适用于只有有限个分类，而且拥有相同F值的记录不是很多的情况。例如网络日志处理和用户分类，用户的总数很多，但是每个用户的事件是有限的，以此分类得到的类别也是有限的。值得一提的是在这种模式下可以在数据传输到Recer之前使用Combiner来去除分类的重复值。

1 class Mapper
2 method Map(null, record [value f, categories [g1, g2,...] )
3 for all category g in [g1, g2,...]
4 Emit(value f, category g)
5
6 class Recer
7 method Initialize
8 H = new AssociativeArray : category -> count
9 method Rece(value f, categories [g1, g2,...])
10 [g1', g2',..] = ExcludeDuplicates( [g1, g2,..] )
11 for all category g in [g1', g2',...]
12 H{g} = H{g} + 1
13 method Close
14 for all category g in H do
15 Emit(category g, count H{g})

应用：日志分析，用户计数
互相关
问题陈述：有多个各由若干项构成的组，计算项两两共同出现于一个组中的次数。假如项数是N，那么应该计算N*N。
这种情况常见于文本分析（条目是单词而元组是句子），市场分析（购买了此物的客户还可能购买什么）。如果N*N小到可以容纳于一台机器的内存，实现起来就比较简单了。
配对法
第一种方法是在Mapper中给所有条目配对，然后在Recer中将同一条目对的计数加和。但这种做法也有缺点：
使用 combiners 带来的的好处有限，因为很可能所有项对都是唯一的
不能有效利用内存

1 class Mapper
2 method Map(null, items [i1, i2,...] )
3 for all item i in [i1, i2,...]
4 for all item j in [i1, i2,...]
5 Emit(pair [i j], count 1)
6
7 class Recer
8 method Rece(pair [i j], counts [c1, c2,...])
9 s = sum([c1, c2,...])
10 Emit(pair[i j], count s)

Stripes Approach（条方法？不知道这个名字怎么理解）
第二种方法是将数据按照pair中的第一项来分组，并维护一个关联数组，数组中存储的是所有关联项的计数。The second approach is to group data by the first item in pair and maintain an associative array (“stripe”) where counters for all adjacent items are accumulated. Recer receives all stripes for leading item i, merges them, and emits the same result as in the Pairs approach.
中间结果的键数量相对较少，因此减少了排序消耗。
可以有效利用 combiners。
可在内存中执行，不过如果没有正确执行的话也会带来问题。
实现起来比较复杂。
一般来说， “stripes” 比 “pairs” 更快

1 class Mapper
2 method Map(null, items [i1, i2,...] )
3 for all item i in [i1, i2,...]
4 H = new AssociativeArray : item -> counter
5 for all item j in [i1, i2,...]
6 H{j} = H{j} + 1
7 Emit(item i, stripe H)
8
9 class Recer
10 method Rece(item i, stripes [H1, H2,...])
11 H = new AssociativeArray : item -> counter
12 H = merge-sum( [H1, H2,...] )
13 for all item j in H.keys()
14 Emit(pair [i j], H{j})

应用：文本分析，市场分析
参考资料：Lin J. Dyer C. Hirst G. Data Intensive Processing MapRece
用MapRece 表达关系模式
在这部分我们会讨论一下怎么使用MapRece来进行主要的关系操作。
筛选（Selection）

1 class Mapper
2 method Map(rowkey key, tuple t)
3 if t satisfies the predicate
4 Emit(tuple t, null)

投影（Projection）
投影只比筛选稍微复杂一点，在这种情况下我们可以用Recer来消除可能的重复值。

1 class Mapper
2 method Map(rowkey key, tuple t)
3 tuple g = project(t) // extract required fields to tuple g
4 Emit(tuple g, null)
5
6 class Recer

⑽ maprece 分区和分组的区别

分区：将map输出的结果按照rece task数量分给不同的rece.
默认算法为map结果的key 进行hash运算，将结果取模。
分组:在分区之后，根据key来进行分组，相同key在同一组