我们为什么需要Map-Reduce？，Map-Reduce?

文章由LinuxBoy分享于2019-03-27 08:03:21热评（600）

我们为什么需要Map-Reduce？，Map-Reduce?

在讨论我们是否真的需要Map-Reduce这一分布式计算技术之前，我们先面对一个问题，这可以为我们讨论这个问题提供一个直观的背景。

问题

这里写图片描述

我们先从最直接和直观的方式出发，来尝试解决这个问题：
先伪一下这个问题：

SELECT COUNT(DISTINCT surname) 
FROM big_name_file

我们用一个指针来关联这个文件.

接着考察每一行的数据，解析出里面的姓氏，这里我们可能需要一个姓氏字典或者对照表，然后我们可以利用最长前缀匹配来解析出姓氏。这很像命名实体识别所干的事。

拿到了姓氏，我们还需要一个链表L，这个链表的每个元素存储两个信息，一个是姓氏或者姓氏的编号，另一个是这个姓氏出现的次数。

在考察每一行的数据时，我们解析出姓氏，然后在链表L中查找这个姓氏对应的元素是否存在，如果存在就将这个元素的姓氏出现次数加一，否则就新增一个元素，然后置这个元素的姓氏出现次数为1。

当所有的行都遍历完毕，链表L的长度就是不同的姓氏的个数出现的次数。

    /**
    *  直接法伪代码
    */
    int distinctCount(file) {
    //将磁盘文件file关联到一个内存中的指针f上
    f <- file;
    //初始化一个链表
    L <- new LinkedList();
    while(true) {
        line <- f.readline();
        if(line == null)
            break;
        //解析出此行的姓氏
        surname <- parse(line);
        //如果链表中没有这个姓氏，就新增一个，如果有，就将这个姓氏的出现次数+1
        L.addOrUpdate(surname,1);
    }
    //链表的长度就是文件中不同姓氏的个数
    return L.size();
}

ok，这个方法在不关心效率和内存空间的情况下是个解决办法。
但是却有一些值得注意的问题：

在进行addOrUpdate操作时，我们需要进行一个find的操作来找到元素是否已在链表中了。对于无序链表来说，我们必须采取逐一比较的方式来实现这个find的语义。

对于上面的考虑，显然我们知道如果能按下标直接找出元素就最好不过了，我们可以在常量时间找出元素并更新姓氏出现的次数。

哈希表法

对于这一点，我们可以采取哈希表来做，采取这个结构，我们可以用常量时间来找到元素并更新。

    int distinctCountWithHashTable(file) {
    //将磁盘文件file关联到一个内存中的指针f上
    f <- file;
    //初始化一个哈希表
    T <- new HashTable();
    while(true) {
        line <- f.readline();
        if(line == null)
            break;
        //解析出此行的姓氏
        surname <- parse(line);
        //如果哈希表中没有这个姓氏，就新增一个，如果有，就将这个姓氏的出现次数+1
        T.addOrUpdate(surname,1);
    }

    //哈希表中实际存储的元素个数就是文件中不同姓氏的个数
    return T.size();
}

假设给定文件是有序的

哈希表法看起来很美，但还是有潜在的问题，如果内存不够大怎么办，哈希表在内存中放不下。这个问题同样存在于直接法中。

想想看，如果这个文件是个排好序的文件，那该多好。
所有重复的姓氏都会连着出现，这样我们只需要标记一个计数器，每次读取一行文本，如果解析出的姓氏和上一行的不同，计数器就增1.
那么代码就像下面这样：

    int distinctCountWithSortedFile(file) {
    //将磁盘文件file关联到一个内存中的指针f上
    f <- file;
    //不同姓氏的计数器，初始为0
    C <- 0;
    //上一行的姓氏
    last_surname <- empty;
    while(true) {
        line <- f.readline();
        if(line == null)
            break;
        //解析出此行的姓氏
        surname <- parse(line);
        //如果和上一行的姓氏不同，计数器加1
        if(!last_surname.equals(surname))
            C++;
        last_surname <- surname;
    }

    return C;
}

遗憾的是，我们并不能保证给定的文件是有序的。但上面方法的优点是可以破除内存空间的限制，对内存的需求很小很小。

那么能不能先排个序呢？
肯定是可以的，那么多排序算法在。但是有了内存空间的限制，能用到的排序算法大概只有位图法和外排了吧。

位图法

假设13亿/32 + 1个int（这里设32位）的内存空间还是有的，那么我们用位图法来做。
位图法很简单，基本上需要两个操作：

    /**
    * 将i编码
    */
    void encode(M,i) {
        (M[i >> 5]) |=  (1 << (i & 0x1F));
    }
    /**
    *将i解码
    */
    int decode(M,i) {
        return (M[i >> 5]) & (1 << (i & 0x1F));
    }

假设我们采取和姓氏字典一样的编号，我们做一个自然升序，那么这个方法就像下面这样：

    int distinctCountWithBitMap(file) {
    //将磁盘文件file关联到一个内存中的指针f上
    f <- file;
    //初始化一个位图结构M，长度为13亿/32 + 1
    M <- new Array();
    //不同姓氏的个数，初始为0
    C <- 0;
    while(true) {
        line <- f.readline();
        if(line == null)
            break;
        //解析出此行的姓氏编号
        surname_index <- parse(line);
        //将姓氏编号编码到位图对应的位上
        encode(M,surname_index);        
    }

    //找出位图中二进制1的个数
    C <- findCountOfOneBits(M);

    return C;
}

ok，一切看起来很完美，但如何有效地找出位图中的二进制1的个数呢？上面使用了一个findCountOfOneBits方法，找出二进制1的个数，好吧，这是另外一个问题，但我们为了完整，可以给出它的一些算法：

int findCountOfOneBits_1(int[] array) {
    int c = 0;
    for(int i = 0 ; i < array.length; i++)
        c += __popcnt(array[i]);
    return c;
}

int findCountOfOneBits_2(int[] array) {
    int c = 0;
    for(int i = 0 ; i < array.length; i++) {
        while(array[i]) {
            array[i] &= array[i] - 1;
            c++;
        }
    }

    return c;
}

int findCountOfOneBits_3(int[] array) {
    int c = 0;
    unsigned int t;
    int e = 0;
    for(int i = 0 ; i < array.length; i++) {
        e = array[i];
        t = e
            - ((e >> 1) & 033333333333)
            - ((e >> 2) & 011111111111);

        t = (t + (t >> 3)) & 030707070707
        c += (t%63);
    }

    return c;
}

上面的算法哪种效率最高呢？老三。

合并法

ok，位图法看起来破除了内存的限制，的确如此吗？如果内存小到连位图都放不下怎么办？
不解决这个问题了！开玩笑~

既然内存严重不足，那么我们只能每次处理一小部分数据，然后对这部分数据进行不同姓氏的个数的统计，用一个{key,count}的结构去维护这个统计，其中key就代表了我们的姓氏，count代表了它出现的次数。

处理完毕一小批数据后，我们需要将统计结果持久化到硬盘，以备最后累计，这牵扯到一个合并的问题。

如何进行有效地合并也值得思索，因为一开始文件内的姓名是无序的，所以不能在最后时刻进行简单合并，因为同一种姓氏可能出现在不同的统计结果分组中，这会使得统计结果出现重复。
所以我们必须对每批统计结果维护一个group结构或者如下的结构：

统计结果1：{{key=赵，count=631}...}
统计结果2：{{key=赵，count=3124}...}
…
统计结果N : {{key=赵，count=9956}...}

这样，我们在最后可以按key进行合并，得出如下的结构：

汇总结果1：{{key=赵，count=20234520}...}
汇总结果2：{{key=王，count=33000091...}
…
汇总结果M：{{key=钱，count=20009323}...}

BTW,数据是瞎编的，我个人并不知道到底哪个姓氏最多。
这样M就是我们不同姓氏的个数。

合并的过程如下图：
这里写图片描述

由于不断地将部分的统计结果合并到硬盘中，这种方式非常类似LSM算法，不同的是，我们对硬盘上中间文件的合并是on-line的，不是off-line的。

分布式法 Map-Reduce

合并法中，显然需要多次的访问硬盘，这有点问题：

如果是机械硬盘，那么磁盘的寻道时间令人头痛。
并且，合并的算法是串行的，我们无法降低摊还寻道代价。

面对内存容量有限的假设，我们可以推广到单机的计算资源有限的场景中来，设想一下，上面所列举的算法中，如果文档是有序的，那么我们仅仅使用极小的内存就可以解决问题，那么我们不需要分布式，也不需要Map-Reduce。

当然，如果我们不仅需要统计不同姓氏的个数，还想知道不同姓氏出现的频率，以研究到底姓王的多还是姓张的多，那么我们需要一些新思路。

如果我们能将姓名数据仔细分组，使得同样的姓氏会出现在同一组中.
然后将这些组分派到不同的计算节点上,由这些节点并行计算出若干个数C1、C2、...、Cn，最终我们的答案就是：n.
而每个姓氏的频率可以表示为：
frequencyi=Ci∑ni=1Ci,其中i是姓氏的编号，Ci表示第i个姓氏的出现的个数。

而对应这种分布式计算模型的，就是Map-Reduce.
一个典型的Map-Reduce模型，大概像下图这样：
这里写图片描述
注：上图来自Search Engines:Information Retrieval In Practice.

对应我们这个问题，伪代码如下：

function Map(file) {
    while(true) {
        line <- file.readline();
        if(line == null)
            break;
        surname <- parse(line);
        count <- 1;
        Emit(surname,count);
    }
}

function Reduce(key,values) {
    C <- 0;
    surname <- key;
    while(!values.empty()) {
        C <- C + values.next();
    }

    Emit(surname,C);
}

使用Map-Reduce技术，不仅可以并行处理姓氏频率，同时也可以应对big、big、big-data（比如全银河系的“人”的姓名）。前提是你有足够的计算节点或者机器。

这里还有一个问题需要注意，就是上面的Reduce算法默认了数据已经按姓氏分组了，这个目标我们依靠Shuffle来完成。

在Shuffle阶段，依靠哈希表来完成group by surname.

在这里，将所有数据按姓氏分组并将每一组分派到一个计算节点上显得有些奢侈，所以如果在机器不足的情况下，可以将分组的粒度变大，比如100个姓氏为一组，然后通过多次的Map-Reduce来获得最终结果。

最后，希望我说明白了为什么我们需要Map-Reduce技术。
同时，不得不承认这个问题的设定是比较尴尬的= _ =，因为在对姓氏的parse阶段，我们用到了一个全姓氏字典，显然这个字典本身(Trie or Hash)可以告诉我们不同姓氏的个数。但如果问题的设定不是全部的姓氏都出现在文件中，或许这篇文章就能起到抛砖引玉的效果，那么其中的过程也值得书写下来。

推荐文章：

我们为什么需要Map-Reduce？，Map-Reduce?