Data-Intensive Text Processing with MapReduce第三章（2）-MapReduce算法设计-3.1局部聚集

3.1局部聚集（local aggregation）

在数据密集的分布式处理环境中，从产生它们的进程到最后消费它们的进程，中间结果的交互是同步中重要的一个方面。在一个集群环境中，除了令人尴尬的并行问题，其它都必须通过网络传输数据。此外，在Hadoop，中间结果是先写到本地磁盘然后再用网络发送出去。因为网络和磁盘因素相对其它因素更加容易成为评价，所以减少中间数据的传输即提高了算法的效率。在MapReduce中，本地中间结果聚集是提高算法效率的一种方法。通过使用combiner（合并器）和通过利用保存不同输入状态的能力，这样做通常能够大体上减少需要从mappers中传输到reducers的键值对数量。

3.1.1 COMBINERS（合并器）和IN-MAPPER COMBINING（MAP内合并）

我们在2.2节中用简单的单词计算例子阐述了本地聚集的各种技术。为了方便，图3.1重现了那个简单的基础算法的伪代码：mapper把每一个term当一条键值对，以term本身作为键，值为1，发送出去；reducers合计部分计数来达到总数。

1: class Mapper

2: method Map(docid a,doc d)

3: for all term t ∈ doc d do

4: Emit(term t, count 1)

1: class Reducer

2: method Reduce(term t; counts [c1, c2, . . .])

3: sum ←0

4: for all count c ∈ counts [c1, c2, . . .] do

5: sum ← sum + c

6: Emit(term t, count sum)

图3.1：MapReduce单词统计算法伪代码（图2.3的再现）

第一个本地聚集的技术是combiner，在2.4节时讨论过。combiner在MapReduce框架中提供一种普通的机制来减少由mappers生成的中间值的数量---我们也可以认为它们是处理mappers输出数据的“迷你reducers”。在这个例子中，合并器统计了每个map任务传来的每个单词个数。这个结果减少了需要通过网络传输的键值对的个数，从集合中所有单词到集合中的部分单词。

1: class Mapper

2: method Map(docid a, doc d)

3: H ←new AssociativeArray

4: for all term t ∈ doc d do

5: H{t} ← H{t} + 1 //Tally counts for entire document

6: for all term t ∈ H do

7: Emit(term t, count H{t})

图3.2：改进型的MapReduce单词统计算法，它使用一个关联数组在每一个文档基础上聚集单词的计数。Reducer和图3.1中的一样。

图3.2展示了改进了的基础算法（mapper被修改，reducer和图3.1的一样）。关联数组（例如java中的map）在mapper中被引进来计算文档中单词的计数：而不是把文档中每个单词的键值对发送出去。考虑到一些词语会经常出现在同一文档中（例如，一篇介绍狗的文章会经常出现“狗”这个单词），这样做可以大量的减少需要发送的键值对的数量，特别是对于长文档来说。

这个基本概念能更进一步，作为图3.3单词统计算法的变种，这个算法的运作严重依赖于2.6节中讲到的hadoop中map和reduce是怎样执行的。回忆，每个map任务都产生一个java mapper对象，它负责处理一堆键值对。在处理键值对之前，mapper的与用户自定代码挂钩的API初始化方法被调用。在这种情况下，我们初始化了一组关联数组来记录单词数。（对每一输入键值对来说）由于它可以在各种Map方法的调用中保持状态，我们可以继续在多个文档间通过关联数组累计部分单词计数，当mapper处理完所有文档后再把键值对发送出去。也就是说，延迟发送中间数据直到运行到伪代码中的close方法。回想起这个API提供一个机会在map方法中所有输入数据中的键值对分配到map任务中执行后执行用户自定代码。

有了这种技术，我们实际上把combiner的功能合并到mapper中。再也不需要运行一个单独的combiner，因为所有本地聚集的情况已经探明。这是一个在MapReduce中比较常见的设计模式，我们称为“in-mapper combining”，以便我们能更方便地在本书中参照这个模式。我们在后面会看到这个模式是怎样应用到不同问题上的。使用这个模式有两个主要好处：

首先，它能够控制本地聚集何时发生和它怎样发生的，相比之下combiner的语义就是在MapReduce的控制下执行。

例如，Hadoop并不保证combiner会用多少次，也就是说它甚至会每次都用。Combiner在执行框架中被作为一个语义维持最优化的提供者，那个会选择使用它，可能会有很多次，或者没有（或者在reduce阶段中）。在某些情况中（虽然不是这个特殊的例子），这种不可预测的情况是无法接受的，所以程序员们经常选择在mappers中执行局部聚集。

其次，使用map内合并（in-mapper combining）通常会比使用combiners更高效。其中一个原因是由于与具体的键值对交互而产生额外的开销。Combiner减少了通过网络传输的中间数据，但实际上没有减少初次发出的键值对数量。图3.2中的算法中，中间键值对仍然在每个文档的基础上生成，只是被combiners减少了些。这个进程涉及不必要的对象的创建和销毁（垃圾处理机制能从容处理），而且还包括对象的序列化与反序列化（当map的输出中间键值对装满内存缓存并需要暂时存放在磁盘时）。相比之下，使用in-mapper combining模式mappers只生成需要通过网络传输给reducer的键值对。

不管怎样也要说说in-mapper combining模式的缺点。首先，它破坏了MapReduce的编程基础，因为状态通过各种输入键值对来保存。最后，它没什么大不了的，因为实事求是地对效率的担忧经常胜过纯理论的，但是实践也是很重要的。通过各种输入实例保存状态意味着算法的行为会依赖于遇到输入键值对时的顺序。这样容易产生潜在的排序依赖错误，它在普通情况下难以调试大的数据集（虽然in-mapper combining的正确性在单词统计的例子上很容易说明）。其次，使用in-mapper combining模式有一个重要的扩展性瓶颈。它严重地依赖于用充足的内存去储存中间结果直到mapper完全处理好输入的所有键值对。在单词统计的例子中，占用内存的多少取决于词典的大少，既然这样那么就有可能mapper会在收集时遇到所有单词。Heap's Law，信息检索中一个熟悉的结果，准确的模拟词典大小的增长对集合大小的影响---稍微出人意料的事实是词典的规模总是在不断扩大。因此，图3.3中的算法将涉及这点，在那些之外的关联数组保持着部分单词计数将不再适用在内存上。

1: class Mapper

2: method Initialize

3: H ← new AssociativeArray

4: method Map(docid a, doc d)

5: for all term t ∈ doc d do

6: H{t} ← H{t} + 1 // Tally counts across documents

7: method Close

8: for all term t ∈ H do

9: Emit(term t, count H{t})

图3.3：演示“in-mapper combining”模式的MapReduce单词统计算法的伪代码，reducer和图3.1一样。

4更多详细资料：注意问题是部分term计数，然而，当集合的大小增加时，其中一个会希望增加输入分割大小值来限制增加的map任务的数量

一个使用in-mapper combining技术时限制内存使用的普遍做法是阻塞键值对和定期清除内存里的数据。这个想法很简单：处理了每N个键值对后就发送这部分结果，而不是所有键值对都处理完后再发送这些中间数据。这可以直接用一个counter变量来跟踪已经处理了的键值对数量。作为一种选择，mapper可以知道自身的内存占用空间和当内存使用不足时清除中间键值对。这两种方法，无论是阻塞值（block size）或内存使用最大值（the memory usage threshold）都要根据经验来用：如果是一个太大的值，mapper会用完内存，但如果值太小，局部聚集的数据又可能丢失。而且，在Hadoop中，物理内存被分配到同一节点的多个同时运行的任务中；这些任务都在竞争着这有限的资源，但因为任务是相互独立的，所以很难有效地去协调使用资源。实际上，受益于不断增加的缓冲值，常常会遇到返回值在执行过程中会逐渐缩小，所以不值得花大力气去寻找设置缓存的最优值（Jeff Dean的个人观点）。

在Mapduce算法中，通过使用局部聚集能增加效率的多少取决于中间值的大小、键值的分布和每个单独任务需要发送键值对的数量。聚集的时机，归根结底是从同一键不同值中得来（不管是否使用combiners或in-mapper combining模式）。在单词统计的例子中，局部聚集之所以高效是因为在一个map任务中很多词语会出现多次。局部聚集也是用来解决减少拖后腿者（看2.3节）即其结果和中间值呈高偏态分布的有效技术。在单词统计的例子中，我们没有过滤频繁出现的词语：因此，没有局部聚集的话，需要计算the个数的reducer相对普通的reducer来说要做很多工作，因此这个reducer就会成为拖后腿者。使用本地聚集（不管使用combiners或in-mapper combining模式），我们大体上减少由频繁出现的词语产生的值，从而缓解减少拖后腿者的问题。

3.1.2 局部聚集算法的正确性

虽然使用combiners可以很好的减少算法的消耗时间，在使用它们时要特别小心。因为在Hadoop中combiner是可选择性的优化手段，算法的正确性并不依赖于combiner的运算能力。在任何MapReduce程序中，reducer的输入键值对类型必须和mapper输出的键值对类型相匹配：这暗示着combiner的输入和输出键值对必须和mapper的输出键值对匹配（像reducer的输入键值对类型那样）。在reduce的计算是交互和协作的例子中，reducer可以作为combiner（像单词统计那个例子）使用（不用修改）。一般来说，combiners和reducers是不可交换的。

想像这样的例子：我们有一个大型数据集，输入键是string类型，输入值是integer类型，我们想计算所有相同键的平均值（rounded to the nearest integer）。现实世界中的例子可能是流行网站的用户操作日志，键代表用户ID，值代表一些用户在网站上的活动，例如在某一会话中停留的时间。这相当于任务会计算基于每一用户的会话平均时间，这将有助于了解用户的特性。图3.4展示了没有用到combiner来完整这个任务的简单算法的伪代码。我们使用一个特别的mapper，它只是把所有的键值对传到reducers（适当地分组和排序）。Reducer记录运行时的最大值和遇到integer的数量。这些信息是用来计算当所有值都处理完后的平均值。平均值最后作为reducer的输出值（输入的string作为键）。

1: class Mapper

2: method Map(string t; integer r)

3: Emit(string t; integer r)

1: class Reducer

2: method Reduce(string t; integers [r1, r2, . . .])

3: sum ← 0

4: cnt ← 0

5: for all integer r ∈ integers [r1, r2, . . .] do

6: sum ← sum + r

7: cnt ← cnt + 1

8: ravg ← sum/cnt

9: Emit(string t, integer ravg)

图 3.4:计算相同key中value的平均值的mapreduce伪代码

这个算法确实可以工作，但是仍然存在像图3.1中的单词统计算法那样的缺点：它需要把所有键值对通过网络从mappers传到reducers，这样非常低效。不像单词统计那例子，在这种情况下reducer不能像combiner那样使用。想象一下如果我们这样做了会发生什么：combiner会计算同一键的任意值的子集，reducer会计算这些结果的平均值。为了更好地说明，我们知道：

Mean(1; 2; 3; 4; 5) ！= Mean(Mean(1; 2)，Mean(3; 4; 5))

一般来说，用一个数据集任意子集的平均值来求平均值和单纯对这个数据集取平均值的结果是不同的。因此这个方法不会得到正确值。

1: class Mapper

2: method Map(string t, integer r)

3: Emit(string t, integer r)

1: class Combiner

2: method Combine(string t, integers [r1, r2, . . .])

3: sum ← 0

4: cnt ← 0

5: for all integer r 2 integers [r1, r2, . . .] do

6: sum ← sum + r

7: cnt ← cnt + 1

8: Emit(string t, pair (sum, cnt)) // 分发总数和计数

1: class Reducer

2: method Reduce(string t, pairs [(s1, c1), (s2, c2) . . .])

3: sum ← 0

4: cnt ← 0

5: for all pair (s, c) ∈ pairs [(s1, c1), (s2, c2) . . .] do

6: sum ← sum + s

7: cnt ← cnt + c

8: ravg ← sum/cnt

9: Emit(string t, integer ravg)

图 3.5:

尝试使用之前提到过的combiner来计算每个key对应value的平均值，combiner中输入和输出键值对类型的不同违背了Mapreduce编程模型的模式。

那么我们怎样合理地利用combiner呢？图3.5做出了一种尝试。Mapper不变，但是我们加入了一个combiner来统计需要计算平均值的数值部分的部分数据。这个combiner会接收每一键值和与之对应的integer值列表，据此计算这些值的总数和遇到integer的数量（即计数）。总数和计数会打包到一个以同一个string作为键的键值对作为combiner的输出发送出去。在reducer中，直到现在，所有我们算法中的键值已经被初始化（变成string，integers）。然而，MapReduce没有禁止使用更复杂的类型，实际上，这代表着一项MapReduce算法设计中的关键技术，我们已经在这章的开头介绍过。我们之后在这本书中还会频繁遇到复杂的键值对问题。

不幸的是，这个算法不能运行。记得combiner必须有着相同的输入输出键值对类型，它必须和mapper的输出类型和reducer的输入类型相匹配。这明显不是这种情况。为了明白在编程模型中为什么必须作出这样的限制，记得combiners只是优化手段，它不会改变算法的正确性。所以让我们移除combiners来看看会发生什么：mapper的输出值类型是integer，所以reducer希望接收一个integer类型的列表来作为值。但reducer实际上是希望传入一个pair类型的列表！算法的正确性可能发生在在mappers输出中运行的combiner上，更确切地说，combiner并不只是运行一次。回忆我们之前讨论过Hadoop并不保证combiners会被调用多少次；可能是零次，一次或多次。这违背了MapReduce的编程模型。

1: class Mapper

2: method Map(string t, integer r)

3: Emit(string t, pair (r, 1))

1: class Combiner

2: method Combine(string t, pairs [(s1, c1), (s2, c2) . . .])

3: sum ← 0

4: cnt ← 0

5: for all pair (s, c) 2 pairs [(s1, c1), (s2, c2) . . .] do

6: sum ← sum + s

7: cnt ← cnt + c

8: Emit(string t, pair (sum, cnt))

1: class Reducer

2: method Reduce(string t, pairs [(s1, c1), (s2, c2) . . .])

3: sum ← 0

4: cnt ← 0

5: for all pair (s, c) 2 pairs [(s1, c1), (s2, c2) . . .] do

6: sum ← sum + s

7: cnt ← cnt + c

8: ravg ← sum/cnt

9: Emit(string t, integer ravg)

图 3.6:计算每个key中value的平均值的mapreduce伪代码。这个代码非常巧妙的使用combiners来实现。

算法的另一个缺点在图3.6中展示，这时候算法是正确的。在mapper，我们发送由整型和1构成的pair值---这相当于一个实例中的部分计数的总和。Combiner分别统计部分总数和部分计数，之后把更新后的总数和计数发送出去。Reducer和combiner类似，唯一不同它是统计最后结果的。实际上，这个算法把一个不能传递的运算（求数字的平均值）转换成能传递的运算（一对数的元素级的总数，在最后有一个额外的分配）。

让我们通过重复之前的实践来证实这个算法的正确性：没有combiners运行时会发生什么？没有combiners的话，mappers会发送pairs（作为值）到reducers。这样的话中间值pairs的数量和输入键值对的数量一样，并且每一个pair值包括一个整型值和1.reducer仍然会达成正确的总数和计数，因此求到的平均值是正确的。现在把combiners加进去：这个算法仍然正确，不管它运行了多长时间，因为combiners只是统计部分总数和计数然后直接传给reducer。值得注意的是虽然combiner的输出键值对类型必须和reducer的输入键值对类型一致，但是reducer可以用不同的数据类型来发送键值对。

1: class Mapper

2: method Initialize

3: S ← new AssociativeArray

4: C ← new AssociativeArray

5: method Map(string t, integer r)

6: S{t} ← S{t} + r

7: C{t} ← C{t} + 1

8: method Close

9: for all term t ∈ S do

10: Emit(term t, pair (S{t},C{t}))

图 3.7:计算关联的value的平均值的伪代码

最后，在图3.7我们展示了一个更加高效的算法来实现in-mapper combining模式。在mapper里面，与键相关的部分总数和计数通过键值对保存在内存上。当所有输入都处理完后才发送中间键值对；与之前例子相同，值是一个由总数和计数组成的对。Reducer和图3.6中的一样。把部分结果统计从combiner移到mapper受到我们在节前讨论过，但在这个例子中，保存中间数据的数据结构在内存的使用上适中，从而使这个变异算法更有吸引力。