首页 / 知识

关于框架：MapReduce的简单解释？

2023-04-13 07:01:00

Simple explanation of MapReduce?

与我的CouchDB问题有关。

谁能用麻木可以理解的术语来解释MapReduce？

一路深入到Map和Reduce的基础知识。

Map是一项功能，可将某种类型的列表中的项目"转换"为另一种类型的项目，并将其放回相同类型的列表中。

假设我有一个数字列表：[1,2,3]，我想对每个数字加倍，在这种情况下，"对每个数字加倍"的函数是函数x = x *2。如果没有映射，我可以编写说一个简单的循环

1 2	A = [1, 2, 3] foreach (item in A) A[item] = A[item] * 2

并且我会有A = [2，4，6]，但是如果没有map函数，我可以编写而不是编写循环

1	A = [1, 2, 3].Map(x => x * 2)

x => x * 2是针对[1,2,3]中的元素执行的函数。发生的情况是该程序获取每个项目，通过使x等于每个项目来对它执行(x => x * 2)，并生成结果列表。

1
2
3

1 : 1 => 1 * 2 : 2
2 : 2 => 2 * 2 : 4
3 : 3 => 3 * 2 : 6

因此，使用(x => x * 2)执行map函数后，您将获得[2，4，6]。

Reduce是一种"收集"列表中的项目并对所有项目执行一些计算，从而将它们减少为单个值的功能。

求和或求平均值是归约函数的所有实例。例如，如果您有一个数字列表，例如[7、8、9]，并且希望对它们进行求和，则可以编写这样的循环

1
2
3

A = [7, 8, 9]
sum = 0
foreach (item in A) sum = sum + A[item]

但是，如果您可以访问reduce函数，则可以这样编写

1 2	A = [7, 8, 9] sum = A.reduce( 0, (x, y) => x + y )

现在，为什么要传递2个参数(0以及带有x和y的函数)有点令人困惑。为了使reduce函数有用，它必须能够取2个项目，计算出某项并将这2个项目"减少"为一个单一值，因此程序可以将每对减少为一个值。

执行如下：

1
2
3
4

result = 0
7 : result = result + 7 = 0 + 7 = 7
8 : result = result + 8 = 7 + 8 = 15
9 : result = result + 9 = 15 + 9 = 24

但是您不想一直都从零开始，因此第一个参数可以让您指定种子值，尤其是第一行result =中的值。

说您想对2个列表求和，则可能看起来像这样：

1
2
3
4
5

A = [7, 8, 9]
B = [1, 2, 3]
sum = 0
sum = A.reduce( sum, (x, y) => x + y )
sum = B.reduce( sum, (x, y) => x + y )

或您更可能在现实世界中找到的版本：

1
2
3
4
5

A = [7, 8, 9]
B = [1, 2, 3]

sum_func = (x, y) => x + y
sum = A.reduce( B.reduce( 0, sum_func ), sum_func )

在数据库软件中这是一件好事，因为有了Map \ reduce支持，您可以使用数据库而无需知道如何将数据存储在数据库中使用数据库，这就是数据库引擎的目的。

您只需要通过向他们提供Map或Reduce函数就可以"告诉"您想要的引擎，然后DB引擎可以找到围绕数据的方式，应用您的函数，并得出您想要的结果想要全部，而您不知道它如何遍历所有记录。

一个数据库可以容纳索引，键，联接和视图以及许多东西，因此通过保护您避免实际存储数据的方式，可以简化代码的编写和维护。

并行编程也是如此，如果您仅指定要对数据执行的操作，而不是实际实现循环代码，则底层基础结构可以"并行化"并为您执行并行并行循环中的功能。

MapReduce是一种可并行处理大量数据的方法，而无需开发人员编写除映射器和reduce函数以外的任何其他代码。

地图功能可将数据输入并取出结果，并将结果保存在障碍中。此功能可以与大量相同的地图任务并行运行。然后可以将数据集缩减为标量值。

因此，如果您将其视为SQL语句

1
2
3
4

SELECT SUM(salary)
FROM employees
WHERE salary > 1000
GROUP by deptname

我们可以使用map来获取薪水> 1000的员工子集
哪个地图将向障碍发射到组大小的存储桶中。

减少将对每个组求和。给您您的结果集。

刚刚从我对Google论文的大学学习笔记中摘下来

收集大量数据

执行某种转换，将每个基准转换为另一种基准

将这些新数据合并为更简单的数据

步骤2是地图。步骤3是减少。

例如，

在道路上的一对压力表上获得两次脉冲之间的时间

根据仪表的距离将这些时间映射为速度

将这些速度降低到平均速度

MapReduce在Map和Reduce之间划分的原因是，可以轻松地并行完成不同部分。 (特别是如果Reduce具有某些数学属性。)

有关MapReduce的复杂但很好的描述，请参阅：Google的MapReduce编程模型-复习(PDF)。

MAP和REDUCE是Lisp的旧功能，源于人类杀死了最后的恐龙。

想象一下，您有一个城市列表，其中包含有关名称，居住人数和城市规模的信息：

1
2
3
4

(defparameter *cities*
'((a :people 100000 :size 200)
(b :people 200000 :size 300)
(c :people 150000 :size 210)))

现在，您可能想要找到人口密度最高的城市。

首先，我们使用MAP创建城市名称和人口密度的列表：

1
2
3
4
5
6
7
8

(map 'list
(lambda (city)
(list (first city)
(/ (getf (rest city) :people)
(getf (rest city) :size))))
*cities*)

=> ((A 500) (B 2000/3) (C 5000/7))

使用REDUCE，我们现在可以找到人口密度最大的城市。

1
2
3
4
5
6
7

(reduce (lambda (a b)
(if (> (second a) (second b))
a
b))
'((A 500) (B 2000/3) (C 5000/7)))

=> (C 5000/7)

结合这两部分，我们得到以下代码：

1
2
3
4
5
6
7
8
9
10

(reduce (lambda (a b)
(if (> (second a) (second b))
a
b))
(map 'list
(lambda (city)
(list (first city)
(/ (getf (rest city) :people)
(getf (rest city) :size))))
*cities*))

让我们介绍一下函数：

1
2
3
4
5
6
7
8
9

(defun density (city)
(list (first city)
(/ (getf (rest city) :people)
(getf (rest city) :size))))

(defun max-density (a b)
(if (> (second a) (second b))
a
b))

然后，我们可以将MAP REDUCE代码编写为：

1
2
3
4

(reduce 'max-density
(map 'list 'density *cities*))

=> (C 5000/7)

它调用MAP和REDUCE(评估是由内而外的)，因此称为map reduce。

让我们以Google论文为例。 MapReduce的目标是能够针对某种算法有效地使用并行处理的处理单元负载。示例如下：您想提取所有单词及其在一组文档中的计数。

典型的实现：

1
2
3
4
5
6

for each document
for each word in the document
get the counter associated to the word for the document
increment that counter
end for
end for

MapReduce的实现：

1
2
3
4
5
6
7
8
9

Map phase (input: document key, document)
for each word in the document
emit an event with the word as the key and the value"1"
end for

Reduce phase (input: key (a word), an iterator going through the emitted values)
for each value in the iterator
sum up the value in a counter
end for

围绕此，您将拥有一个主程序，该程序将以"拆分"的形式对文档集进行分区，这将在Map阶段并行处理。工作者将发出的值写入该工作者专用的缓冲区中。然后，主程序在得知缓冲区已准备好进行处理时，委派其他工作人员执行Reduce阶段。

实际上，每个工作程序输出(作为Map或Reduce工作程序)都是存储在分布式文件系统(Google的GFS)或CouchDB的分布式数据库中的文件。

有关MapReduce的真正简单，快速和"傻瓜式"介绍，请访问：http://www.marcolotz.com/?p=67

好的。

发布其中的一些内容：

好的。

首先，为什么最初创建MapReduce？

好的。

基本上，Google需要一种使大型计算任务易于并行化的解决方案，以允许将数据分发到通过网络连接的多台计算机中。除此之外，它还必须以透明的方式处理机器故障并管理负载平衡问题。

好的。

MapReduce的真正优势是什么？

好的。

可能有人会说MapReduce魔术基于Map and Reduce函数应用程序。我必须承认伴生，我非常不同意。使MapReduce如此流行的主要功能是它的自动并行化和分发功能以及简单的界面。这些因素加上对大多数错误的透明故障处理，使得此框架如此受欢迎。

好的。

在纸上多一点深度：

好的。

MapReduce最初是在Google的一篇论文中提到的(Dean＆Ghemawat，2004 –链接在此)，是一种使用并行方法和商用计算机集群在大数据中进行计算的解决方案。与用Java编写的Hadoop相比，Google的框架使用C ++编写。该文档描述了使用大型数据集上的函数编程中的Map和Reduce函数使用并行框架的行为。

好的。

在此解决方案中，将有两个主要步骤-称为Map和Reduce-，而在第一和第二个步骤之间有一个可选步骤-称为合并。映射步骤将首先运行，在输入键值对中进行计算并生成新的输出键值。必须记住，输入键值对的格式不必一定要与输出格式对匹配。 Reduce步骤将组装同一键的所有值，并对其执行其他计算。结果，这最后一步将输出键值对。 MapReduce最简单的应用程序之一是实现字数统计。

好的。

下面给出了此应用程序的伪代码：

好的。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

map(String key, String value):

// key: document name
// value: document contents
for each word w in value:
EmitIntermediate(w,"1");

reduce(String key, Iterator values):

// key: a word
// values: a list of counts
int result = 0;
for each v in values:
result += ParseInt(v);
Emit(AsString(result));

可以注意到，映射读取记录中的所有单词(在这种情况下，记录可以是一行)，并发出该单词作为键，并发出数字1作为值。
稍后，reduce将对同一键的所有值进行分组。让我们举个例子：假设"房子"一词在记录中出现了3次。减速器的输入为[house，[1,1,1]]。在化简器中，它将对密钥库的所有值求和，并给出以下密钥值作为输出：[house，[3]]。

好的。

这是在MapReduce框架中的外观图：

好的。

Image from the Original MapReduce Google paper

好的。

作为MapReduce应用程序的其他一些经典示例，可以说：

好的。

URL访问次数计数

好的。

反向Web链接图

好的。

分布式Grep

好的。

每个主机的术语向量

好的。

为了避免过多的网络流量，本文描述了框架应如何尝试维护数据局部性。这意味着它应始终尝试确保运行Map作业的计算机上的数据位于其内存/本地存储中，避免从网络中获取数据。为了通过放置映射器减少网络，使用了前面描述的可选组合器步骤。组合器在将给定机器中的映射器的输出发送到Reducers(可能位于另一台机器中)之前，会对它们的输出执行计算。

好的。

该文档还描述了框架元素在出现故障时应如何表现。在本文中，这些元素称为工作人员和主管。在开源实现中，它们将被划分为更具体的元素。
由于Google仅在本文中描述了该方法，而未发布其专有软件，因此创建了许多开源框架以实现该模型。例如，可以说Hadoop或MongoDB中有限的MapReduce功能。

好的。

运行时应注意非专家程序员的细节，例如对输入数据进行分区，在大型机器上安排程序执行，处理机器故障(当然是透明的)以及管理机器间通信。有经验的用户可以调整这些参数，因为输入数据将在工作人员之间进行分区。

好的。

关键概念：

好的。

？容错：必须容忍机器故障。为了执行此操作，主机定期对工人执行ping操作。如果主服务器在一定时间内未收到给定工作人员的响应，则主服务器将工作定义为该工作人员失败。在这种情况下，有故障的工作人员完成的所有地图任务都将被丢弃，并交给其他可用的工作人员。如果工作人员仍在处理地图或简化任务，也会发生类似情况。请注意，如果工作程序已经完成了其缩减部分，则所有计算都将在其失败时完成，并且无需重置。作为主要故障点，如果主服务器失败，则所有作业都会失败。因此，可以为主机定义定期检查点，以保存其数据结构。在最后一个检查点和主服务器故障之间发生的所有计算都将丢失。

好的。

？局部性：为了避免网络流量，框架试图确保所有输入数据对于将在其上执行计算的计算机本地可用。在原始说明中，它使用Google File System(GFS)，复制因子设置为3，块大小为64 MB。这意味着相同的64 MB块(在文件系统中组成一个文件)将在三台不同的计算机上具有相同的副本。主机知道这些块在哪里，并尝试在该机器上计划地图作业。如果失败，则主服务器尝试在任务输入数据的副本附近分配一台计算机(即，工作计算机位于数据机的同一机架中)。

好的。

任务粒度：假设每个映射阶段均分为M个部分，并且每个Reduce阶段均分为R个部分，则理想的情况是M和R比工作机器的数量大得多。这是由于以下事实：执行许多不同任务的工作人员可以改善动态负载平衡。除此之外，在工作人员失败的情况下，它还提高了恢复速度(因为它完成的许多映射任务可以分布在所有其他机器上)。

好的。

备份任务：有时，Map或Reducer工作者的行为可能比群集中其他工作者的行为慢得多。这可以保留总处理时间，并使它等于该慢速计算机的处理时间。原始文件描述了一种称为备份任务的替代方法，该替代任务是在MapReduce操作接近完成时由主服务器调度的。这些是由正在进行的任务的主机计划的任务。因此，当主数据库或备份数据库完成时，MapReduce操作完成。

好的。

计数器：有时可能希望对事件的发生进行计数。因此，在创建位置计数。每个工作程序中的计数器值会定期传播到主服务器。然后，主服务器进行汇总(是的。看起来像Pregel聚合器来自此位置)；成功映射并减少任务的计数器值，并在MapReduce操作完成时将其返回给用户代码。在主状态中还有一个可用的当前计数器值，因此观看此过程的人可以跟踪其行为方式。

好的。

好吧，我想基于上述所有概念，Hadoop对您来说将是小菜一碟。如果您对原始MapReduce文章或任何相关问题有任何疑问，请告诉我。

好的。

好。

如果您熟悉Python，则以下是对MapReduce的最简单的解释：

1
2
3
4
5
6
7
8
9
10

In [2]: data = [1, 2, 3, 4, 5, 6]
In [3]: mapped_result = map(lambda x: x*2, data)

In [4]: mapped_result
Out[4]: [2, 4, 6, 8, 10, 12]

In [10]: final_result = reduce(lambda x, y: x+y, mapped_result)

In [11]: final_result
Out[11]: 42

了解如何分别处理原始数据的每个部分，在这种情况下，将它们乘以2(MapReduce的地图部分)。基于mapped_result，我们得出的结果是42(MapReduce的缩小部分)。

该示例的一个重要结论是，每个处理块都不依赖于另一个块。例如，如果thread_1映射[1, 2, 3]，并且thread_2映射[4, 5, 6]，则两个线程的最终结果仍将是[2, 4, 6, 8, 10, 12]，但为此我们将处理时间减少了一半。对于reduce操作可以说相同，这是MapReduce在并行计算中如何工作的本质。

我不想听起来陈旧，但这对我有很大帮助，这很简单：