MapReduce-大规模数据集分布式并行运算编程模型

    本文转载自CSDN博客，纯为技术资料备份！

    MapReduce的名字源于函数式编程模型中的两项核心操作：Map和Reduce。也许熟悉Functional Programming(FP)的人见到这两个词会倍感亲切。因为Map和Reduce这两个术语源自Lisp语言和函数式编程。Map是把一组数据一对一的映射为另外的一组数据，其映射的规则由一个函数来指定。Reduce是对一组数据进行归约，这个归约的规则由一个函数指定。Map是一个把数据分开的过程，Reduce则是把分开的数据合并的过程。如Hadoop的wordcount例子：用Map把[one,word,one,dream]进行映射就变成了[{one,1}, {word,1}, {one,1}, {dream,1}]，再用Reduce把[{one,1}, {word,1}, {one,1}, {dream,1}]归约变成[{one,2}, {word,1}, {dream,1}]的结果集。

 

    对数组里的每个元素进行相同操作的一段代码：

a = [1, 2, 3];
for (i = 0; i < a.length; i++){
    a[i] = a[i] * 2;
}
for (i = 0; i < a.length; i++){
    print(a[i]);
}

 

   常常要对数组里的所有元素做同一件事情，因此你可以写个这样的函数：   

function map(fn, a){
     for (i = 0; i < a.length; i++){
             fn(a[i]);
    }
}

 

   现在可以把上面的东西改成：

map(function(x) { return x * 2; }, a);
map(print, a);

 

    另一个常见的任务是将数组内的所有元素按照某种方式汇总起来：

function sum(a){
    s = 0;
    for (i = 0; i < a.length; i++){
        s += a[i];
    }
    return s;
}

function join(a){
    s = "";
    for (i = 0; i < a.length; i++){
         s = s .. a[i]; // ..是字符串连接操作符
    }
    return s;
}

print(sum([1,2,3]));
print(join(["a","b","c"]));

    注意sum和join长得很像，你也许想把它们抽象为一个将数组内的所有元素按某种算法汇总起來的泛型函数：

function reduce(fn, a, init){
   s = init;
   for (i = 0; i < a.length; i++){
      s = fn(s, a[i]);
   }
   return s;
}

    这样sum和join就变成下面的样子了：

function sum(a){
    return reduce(function(a, b) { return a + b; }, a, 0 );
}

function join(a){
    return reduce(function(a, b) { return a .. b; }, a, "" );
}

 

    让我们看回map函数。当你要对数组内的每个元素做一些事，你很可能不在乎哪个元素先做。无论由第一个元素开始执行，还是是由最后一个元素开始执行，你的结果都是一样的。这样如果你手头上有2个CPU，你可以写段代码，使它们各自处理1/2的元素，于是乎map快了两倍。设想你在全球有千千万万台服务器，恰好你有一个真的很大很大的数组，现在你可以在几千台服务器上同时执行map，让每台服务器都来解决同一个问题的一小部分。

     Map的定义：

Map, written by the user, takes an input pair and produces a set of intermediate key/value pairs. The MapReduce library groups together all intermediate values associated with the same intermediate key I and passes them to the Reduce function.

     Reduce的定义：

The Reduce function, also written by the user, accepts an intermediate key I and a set of values for that key. It merges together these values to form a possibly smaller set of values. Typically just zero or one output value is produced per Reduce invocation. The intermediate values are supplied to the user’s reduce function via an iterator. This allows us to handle lists of values that are too large to fit in memory.

 

    MapReduce论文中给出了这样一个例子：在一个文档集合中统计每个单词出现的次数。Map操作的输入是每一篇文档，将输入文档中每一个单词的出现输出到中间文件中去。

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

    比如我们有两篇文档，内容分别是

        A － "I love programming"

        B － "I am a blogger, you are also a blogger"

    A文档经过Map运算后输出的中间文件将会是：

        I,1

        love,1

        programming,1

    B文档经过Map运算后输出的中间文件将会是：

        I,1

       am,1

       a,1

       blogger,1

       you,1

       are,1

       also,1

       a,1

       blogger,1

    Reduce操作的输入是单词和出现次数的序列。用上面的例子来说，就是 (I, [1, 1]), (love, [1]), (programming, [1]), (am, [1]), (a, [1,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));

    最终结果是：("I", "2"), ("a", "2"), ……

    实际的执行顺序是：

        MapReduce Library将Input分成M份。这里的Input Splitter也可以是多台机器并行Split。

        Master将M份Job分给Idle状态的M个worker来处理；

        对于输入中的每一个<key, value> pair 进行Map操作，将中间结果Buffer在Memory里；

        定期的（或者根据内存状态），将Buffer中的中间信息Dump到本地磁盘上，并且把文件信息传回给Master（Master需要把这些信息发送给Reduce worker）。这里最重要的一点是，在写磁盘的时候，需要将中间文件做Partition（比如R个）。拿上面的例子来举例，如果把所有的信息存到一个文件，Reduce worker又会变成瓶颈。我们只需要保证相同Key能出现在同一个Partition里面就可以把这个问题分解。

         R个Reduce worker开始工作，从不同的Map worker的Partition那里拿到数据（read the buffered data from the local disks of the map workers），用key进行排序（如果内存中放不下需要用到外部排序 – external sort）。很显然，排序（或者说Group）是Reduce函数之前必须做的一步。 这里面很关键的是，每个Reduce worker会去从很多Map worker那里拿到X(0<X<R) Partition的中间结果，这样，所有属于这个Key的信息已经都在这个worker上了。

         Reduce worker遍历中间数据，对每一个唯一Key，执行Reduce函数（参数是这个key以及相对应的一系列Value）。

         执行完毕后，唤醒用户程序，返回结果（最后应该有R份Output，每个Reduce Worker一个）。

     可见，这里的分（Divide）体现在两步，分别是将输入分成M份，以及将Map的中间结果分成R份。将输入分开通常很简单，Map的中间结果通常用”hash(key) mod R”这个结果作为标准，保证相同的Key出现在同一个Partition里面。当然，使用者也可以指定自己的Partition Function，比如，对于Url Key，如果希望同一个Host的URL出现在同一个Partition，可以用”hash(Hostname(urlkey)) mod R”作为Partition Function。

     对于上面的例子来说，每个文档中都可能会出现成千上万的 ("the", 1)这样的中间结果，琐碎的中间文件必然导致传输上的损失。因此，MapReduce还支持用户提供Combiner Function。这个函数通常与Reduce Function有相同的实现，不同点在于Reduce函数的输出是最终结果，而Combiner函数的输出是Reduce函数的某一个输入的中间文件。