🗺️ 01 · MapReduce：从原理到手写词频统计

%matplotlib inline
import matplotlib
import matplotlib.pyplot as plt
matplotlib.rcParams['font.sans-serif'] = ['Microsoft YaHei', 'SimHei', 'PingFang SC', 'Noto Sans CJK SC', 'Arial Unicode MS']
matplotlib.rcParams['axes.unicode_minus'] = False
import numpy as np
import pandas as pd
import sqlite3
import os
rng = np.random.default_rng(42)

print('''1) Map：把每个输入记录变成 (key, value) 中间结果，可并行；
2) Shuffle：系统按 key 排序并分组（相同 key 发给同一 Reduce）；
3) Reduce：对一个 key 的所有 value 做聚合，输出最终结果。

核心价值：把"不可并行"的聚合问题拆成"可并行"的 map + 局部聚合。
''')

1) Map：把每个输入记录变成 (key, value) 中间结果，可并行；
2) Shuffle：系统按 key 排序并分组（相同 key 发给同一 Reduce）；
3) Reduce：对一个 key 的所有 value 做聚合，输出最终结果。

核心价值：把"不可并行"的聚合问题拆成"可并行"的 map + 局部聚合。

# 模拟 3 个数据分片，每片若干行单词
splits = [
    ['apple banana apple', 'cherry apple'],
    ['banana banana cherry', 'apple'],
    ['cherry cherry banana apple', 'apple cherry'],
]
for i, s in enumerate(splits):
    print(f'分片{i}: {s}')
print('\n➡️ 真实场景：每个分片在（虚拟）不同机器上并行 Map')

分片0: ['apple banana apple', 'cherry apple']
分片1: ['banana banana cherry', 'apple']
分片2: ['cherry cherry banana apple', 'apple cherry']

➡️ 真实场景：每个分片在（虚拟）不同机器上并行 Map

def mapper(line):
    out = []
    for w in line.lower().split():
        out.append((w, 1))
    return out

intermediate = []
for i, split in enumerate(splits):
    partition_out = []
    for line in split:
        partition_out += mapper(line)
    intermediate += partition_out
    print(f'分片{i} Map 输出: {partition_out}')
print(f'\n全部中间结果 {len(intermediate)} 条')
print('➡️ Map 输出也称 intermediate records，key=词，value=1')

分片0 Map 输出: [('apple', 1), ('banana', 1), ('apple', 1), ('cherry', 1), ('apple', 1)]
分片1 Map 输出: [('banana', 1), ('banana', 1), ('cherry', 1), ('apple', 1)]
分片2 Map 输出: [('cherry', 1), ('cherry', 1), ('banana', 1), ('apple', 1), ('apple', 1), ('cherry', 1)]

全部中间结果 15 条
➡️ Map 输出也称 intermediate records，key=词，value=1

# Shuffle = 排序（按 key） + 分组（相同 key 的 value 集合到一起）
intermediate_sorted = sorted(intermediate, key=lambda kv: kv[0])
grouped = {}
for k, v in intermediate_sorted:
    grouped.setdefault(k, []).append(v)
print('排序后（前 8 条）:', intermediate_sorted[:8])
print('\n分组结果:', {k: len(v) for k, v in grouped.items()})
print('➡️ 面试点：Shuffle 是 MR 最耗时的环节（排序+网络传输），'
      '优化思路：map 端预聚合（combiner）、压缩中间数据')

排序后（前 8 条）: [('apple', 1), ('apple', 1), ('apple', 1), ('apple', 1), ('apple', 1), ('apple', 1), ('banana', 1), ('banana', 1)]

分组结果: {'apple': 6, 'banana': 4, 'cherry': 5}
➡️ 面试点：Shuffle 是 MR 最耗时的环节（排序+网络传输），优化思路：map 端预聚合（combiner）、压缩中间数据

result = {k: sum(v) for k, v in grouped.items()}
total = sum(result.values())
print('词频统计结果:', dict(sorted(result.items())))
print('总单词数:', total)

# 对照 pandas 验证
df = pd.DataFrame([w for s in splits for line in s for w in line.lower().split()], columns=['word'])
expected = df['word'].value_counts().sort_index().to_dict()
assert result == expected, 'MR 结果与 pandas 对照不一致'
print('✅ 与 pandas value_counts 对照完全一致')
total_words = sum(len(line.split()) for s in splits for line in s)
print('数据总单词数:', total_words)
assert total_words == 15
print('✅ MapReduce 逻辑正确')

词频统计结果: {'apple': 6, 'banana': 4, 'cherry': 5}
总单词数: 15
✅ 与 pandas value_counts 对照完全一致
数据总单词数: 15
✅ MapReduce 逻辑正确

# 可视化：词频统计结果
fig, ax = plt.subplots(figsize=(7.5, 4.2))
words_sorted = sorted(result.items(), key=lambda kv: -kv[1])
ks = [k for k, _ in words_sorted]; vs = [v for _, v in words_sorted]
bars = ax.bar(ks, vs, color=['#2F5597', '#ED7D31', '#548235', '#C00000'][:len(ks)], width=0.55)
for b, v in zip(bars, vs):
    ax.text(b.get_x() + b.get_width() / 2, v + 0.06, str(v), ha='center', fontsize=12, weight='bold')
ax.set_ylabel('出现次数'); ax.set_title('MapReduce 词频统计结果（15 个单词，与 pandas 交叉验证一致）')
ax.set_ylim(0, max(vs) + 1.2)
plt.show()

# 平均分：Map 出 (班级, 分数)，Reduce 求和并计数
records = [('一班', 85), ('二班', 92), ('一班', 78), ('三班', 88), ('二班', 95), ('一班', 90), ('三班', 70)]
grouped_scores = {}
for cls, sc in records:
    grouped_scores.setdefault(cls, []).append(sc)
avg = {cls: round(np.mean(vs), 2) for cls, vs in grouped_scores.items()}
print('各班平均分:', avg)

# TopN：全局 top2（跨分片，可用分区+归并）
top2 = sorted((v, k) for k, v in result.items())[-2:]
print('词频 Top2:', [(k, v) for v, k in top2[::-1]])
assert len(avg) == 3
print('✅ 变体任务完成')

各班平均分: {'一班': np.float64(84.33), '二班': np.float64(93.5), '三班': np.float64(79.0)}
词频 Top2: [('apple', 6), ('cherry', 5)]
✅ 变体任务完成

# 验证：均值任务必须传 (sum, count) 而不是局部均值
records_c = [('一班', 85), ('一班', 78), ('一班', 90), ('二班', 92), ('二班', 95)]
acc = {}
for cls, sc in records_c:
    s, c = acc.get(cls, (0, 0))
    acc[cls] = (s + sc, c + 1)
correct = {cls: round(s / c, 2) for cls, (s, c) in acc.items()}
wrong = {'一班': round(np.mean([85, 78]) , 2)}  # 错误示范：局部平均后直接再平均
print('正确(传和与计数):', correct, ' | 一班真实均值 =', round(np.mean([85, 78, 90]), 2))
print('错误(传局部均值):', wrong, ' -> 与真实值不符')
assert correct['一班'] == 84.33

正确(传和与计数): {'一班': 84.33, '二班': 93.5}  | 一班真实均值 = 84.33
错误(传局部均值): {'一班': np.float64(81.5)}  -> 与真实值不符

print('''1) 磁盘 IO 重：中间结果落盘，迭代计算慢；
   -> 被 Spark（内存计算）替代，仅少数批任务保留；
2) 编程模型受限：复杂 DAG/迭代算法难表达；
3) 实时性差：分钟级延迟起步；
4) 大量 shuffle 排序开销；
5) 适用场景：超大单遍聚合（离线日志统计），如今多由 Hive/Spark SQL 语法糖承接。
''')

1) 磁盘 IO 重：中间结果落盘，迭代计算慢；
   -> 被 Spark（内存计算）替代，仅少数批任务保留；
2) 编程模型受限：复杂 DAG/迭代算法难表达；
3) 实时性差：分钟级延迟起步；
4) 大量 shuffle 排序开销；
5) 适用场景：超大单遍聚合（离线日志统计），如今多由 Hive/Spark SQL 语法糖承接。