MapReduce——2004年，Google 如何让上千台普通计算机共同处理海量数据？

(单词, 1)

cat dog cat

(cat, 1)
(dog, 1)
(cat, 1)

(dog, 1)
(bird, 1)

(cat, 1)
(bird, 1)
(dog, 1)

cat → [1,1,1]

dog → [1,1,1]

sum(values)

key
values_for_this_key

hash("cat") % 3 = 1

hash("dog") % 3 = 2

(cat,1)
(bird,1)
(cat,1)
(bird,1)
(cat,1)

(bird,1)
(bird,1)
(cat,1)
(cat,1)
(cat,1)

reduce("bird", [1,1])
reduce("cat", [1,1,1])

(cat, 1)
(cat, 1)
(cat, 1)
...

(cat, 100000)

(sum, count)

def map_function(document):
    return count_words(document)

def map_function(order):
    payment_api.charge(
        order.user_id,
        order.amount
    )

    return ("charged", 1)

import re
from collections import defaultdict


documents = {
    "doc1": "cat dog cat",
    "doc2": "dog bird",
    "doc3": "cat bird dog",
}


def map_words(document_id, text):
    """把文档变成单词计数键值对"""
    for word in re.findall(r"[a-z]+", text.lower()):
        yield word, 1


def local_combine(pairs):
    """在一个 Map 任务内部提前合并"""
    counts = defaultdict(int)

    for key, value in pairs:
        counts[key] += value

    return list(counts.items())


def shuffle(map_outputs):
    """把所有 Map 输出按照 key 归组"""
    grouped = defaultdict(list)

    for task_output in map_outputs:
        for key, value in task_output:
            grouped[key].append(value)

    return grouped


def reduce_sum(key, values):
    """将相同 key 的值汇总"""
    return key, sum(values)


def run_mapreduce(documents, use_combiner=True):
    map_outputs = []

    # Map
    for doc_id, text in documents.items():
        pairs = list(map_words(doc_id, text))

        if use_combiner:
            pairs = local_combine(pairs)

        map_outputs.append(pairs)

    # Shuffle
    grouped = shuffle(map_outputs)

    # Reduce
    result = {}

    for key in sorted(grouped):
        word, count = reduce_sum(
            key,
            grouped[key]
        )

        result[word] = count

    return result


print(run_mapreduce(documents))

{'bird': 2, 'cat': 3, 'dog': 3}

outputs = [
    list(map_words("doc1", documents["doc1"])),
    list(map_words("doc1", documents["doc1"])),
]

# 同一个逻辑任务可以被执行多次
attempts = [
    ("doc1", "W1", list(map_words(
        "doc1", documents["doc1"]
    ))),
    ("doc1", "W3", list(map_words(
        "doc1", documents["doc1"]
    ))),
    ("doc2", "W2", list(map_words(
        "doc2", documents["doc2"]
    ))),
    ("doc3", "W4", list(map_words(
        "doc3", documents["doc3"]
    ))),
]

# 教学模拟：Master 只接受每个逻辑任务
# 的一个有效完成结果
accepted = {}

for task_id, worker, output in attempts:
    if task_id not in accepted:
        accepted[task_id] = output

grouped = shuffle(accepted.values())

result = {
    key: sum(values)
    for key, values in grouped.items()
}

print(result)
# {'cat': 3, 'dog': 3, 'bird': 2}

cat cat dog

(cat, 1)
(cat, 1)
(dog, 1)