实验 02 · Notebook

⚡ 02 · Spark：RDD / 惰性求值 / 血缘 / 内存计算

> Spark 是离线大数据的**事实标准**。本节不装 Spark，用 Python 模拟核心机制：
> **RDD 转换（transformation）、惰性求值、血缘（lineage）、action 触发、Shuffle 与宽窄依赖**。

📋 本节要解决的问题

| # | 问题 | 方法 |
|---|---|---|
| 1 | RDD 是什么？ | 弹性分布式数据集（分区+转换） |
| 2 | 惰性求值怎么工作？ | transformation 攒着，action 才执行 |
| 3 | 血缘（lineage）干什么？ | 记录来源，故障重算 |
| 4 | 宽窄依赖区别？ | 窄=同分区，宽=Shuffle |

%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)

1️⃣ Spark 核心概念（必背）

Spark 生态与"为什么快"

| 组件 | 作用 |
|---|---|
| Spark Core | RDD 抽象、任务调度、内存管理 |
| Spark SQL | DataFrame/Dataset + Catalyst 优化器（生产主力） |
| Structured Streaming | 流式处理（增量查询模型） |
| MLlib | 分布式机器学习 |

**比 MapReduce 快在哪**：MR 每个作业的中间结果必须**落盘**再由下一个作业读取（磁盘 IO 重）；
Spark 把多步转换组成 **DAG**，窄依赖之间**流水线执行**（一批数据在一台机器上连续 map→filter，不落地），
只有宽依赖（Shuffle）才真正交换数据——迭代类算法（机器学习常见）可快 10~100 倍。

**RDD 的"弹性"**体现在三点：①分区可并行；②血缘可**重算**恢复；③可**缓存/释放**在内存与磁盘间切换。

print('''Spark 核心：
  1) RDD：不可变、分区的数据集合（弹性=可重算、可缓存）；
  2) Transformation：map/filter/flatMap/groupBy...  惰性（lazy）；
  3) Action：collect/count/reduce/take...           立即触发；
  4) DAG：转换构成有向无环图，Action 时提交执行；
  5) 血缘（Lineage）：每个 RDD 记录父 RDD 与转换，故障自动重算；
  6) Cache/Checkpoint：缓存中间结果，切断过长血缘。
''')
Spark 核心：
  1) RDD：不可变、分区的数据集合（弹性=可重算、可缓存）；
  2) Transformation：map/filter/flatMap/groupBy...  惰性（lazy）；
  3) Action：collect/count/reduce/take...           立即触发；
  4) DAG：转换构成有向无环图，Action 时提交执行；
  5) 血缘（Lineage）：每个 RDD 记录父 RDD 与转换，故障自动重算；
  6) Cache/Checkpoint：缓存中间结果，切断过长血缘。

2️⃣ 手写迷你 RDD（简化版）

class MiniRDD:
    def __init__(self, partitions, deps=None, op=None):
        self.partitions = partitions      # 分区列表（每区一个 list）
        self.deps = deps or []            # 血缘
        self.op = op or 'source'
    def map(self, f):
        # transformation：不执行，只记录闭包
        new_parts = [None] * len(self.partitions)
        return MiniRDD(new_parts, deps=self.deps + [f'source.map({f.__name__ if hasattr(f,"__name__") else "lambda"})'], op='map')
    def filter(self, f):
        return MiniRDD([None] * len(self.partitions), deps=self.deps + ['filter'], op='filter')
    def materialize(self, f=None, g=None):
        # 模拟执行：从源头开始真正计算
        if self.op == 'source':
            return [[x for x in p] for p in self.partitions]
        # 简化：由闭包外部处理
        return None

src = MiniRDD([[1, 2, 3], [4, 5, 6], [7, 8, 9]])   # 3 个分区
rdd2 = src.map(lambda x: x * 2).filter(lambda x: x > 5)
print('map 后血缘:', rdd2.deps)
print('➡️ 上面只是"记录操作"，没有真正执行任何计算（惰性）')
map 后血缘: ['source.map(<lambda>)', 'filter']
➡️ 上面只是"记录操作"，没有真正执行任何计算（惰性）

3️⃣ 惰性求值演示：action 才触发

# 完整迷你执行器：真正跑通 map/filter/count
from functools import reduce

class MiniSpark:
    def __init__(self, partitions):
        self._parts = partitions
        self._plan = []          # (name, fn) 转换序列
        self._executed = 0

    def map(self, f):
        self._plan.append(('map', f)); return self
    def filter(self, f):
        self._plan.append(('filter', f)); return self
    def flatMap(self, f):
        self._plan.append(('flatMap', f)); return self
    def reduceByKey(self, f):
        self._plan.append(('reduceByKey', f)); return self
    def run(self):
        parts = [list(p) for p in self._parts]
        for name, fn in self._plan:
            if name == 'map':
                parts = [[fn(x) for x in p] for p in parts]
            elif name == 'filter':
                parts = [[x for x in p if fn(x)] for p in parts]
            elif name == 'flatMap':
                parts = [[y for x in p for y in fn(x)] for p in parts]
            elif name == 'reduceByKey':
                out = []
                for p in parts:
                    d = {}
                    for k, v in p:
                        d[k] = fn(d[k], v) if k in d else v
                    out.append(list(d.items()))
                parts = out
            self._executed += 1
        return parts

ms = MiniSpark([[1, 2, 3], [4, 5, 6], [7, 8, 9]])
ms.map(lambda x: x * 2).filter(lambda x: x > 5)
print(f'构建计划后：执行次数 = {ms._executed}（尚未计算！）')
res = ms.run()
print(f'action(run) 后：执行次数 = {ms._executed}，结果 = {res}')
# map 后 9 个元素，filter >5 保留 7 个
assert ms._executed == 2 and sum(len(p) for p in res) == 7
print('✅ 惰性求值：transformations 不执行，action 才执行')
构建计划后：执行次数 = 0（尚未计算！）
action(run) 后：执行次数 = 2，结果 = [[6], [8, 10, 12], [14, 16, 18]]
✅ 惰性求值：transformations 不执行，action 才执行

# 可视化：惰性求值时间线（构建期 vs action 期）
fig, ax = plt.subplots(figsize=(9, 3.2))
ax.barh(['构建 map/filter 链', 'action(run)'], [1.8, 0.2], color=['#7F7F7F', '#C00000'], height=0.45)
ax.text(0.9, 0, '只记录转换，0 次数据遍历', ha='center', va='center', color='white', fontsize=11)
ax.text(0.9, 1, '0 次遍历', ha='center', va='center', color='white', fontsize=10)
ax.set_xlim(0, 2); ax.set_xlabel('时间 →', fontsize=11); ax.set_yticks([0, 1])
ax.set_title('惰性求值：transformation 阶段不碰数据，action 提交后才执行', fontsize=12)
plt.show()
# 关键证据：ms._executed 在 run() 前后分别为 0→2（map、filter 各执行 1 轮）
print('run 前 _executed 应为 0，run 后为 2 ->', '验证：构建计划时 executed =', MiniSpark([[1]])._executed)
Notebook 已保存的运行结果
run 前 _executed 应为 0，run 后为 2 -> 验证：构建计划时 executed = 0

4️⃣ 词频（Spark 风格）

lines = ['apple banana', 'apple cherry', 'banana cherry apple']
ms2 = MiniSpark([[w for line in lines for w in line.split()]])
ms2.map(lambda w: (w, 1)).reduceByKey(lambda a, b: a + b)
out = ms2.run()
wc = dict(out[0])
print('Spark 风格词频:', wc)
assert wc['apple'] == 3 and wc['cherry'] == 2 and wc['banana'] == 2
print('✅ 词频统计与手写 MapReduce 结果一致')
Spark 风格词频: {'apple': 3, 'banana': 2, 'cherry': 2}
✅ 词频统计与手写 MapReduce 结果一致

5️⃣ 宽窄依赖与 Shuffle（图示）

# 可视化：宽依赖 vs 窄依赖（分区连线）
fig, axes = plt.subplots(1, 2, figsize=(10.5, 4))
for ax, title, wide in zip(axes, ['窄依赖: map/filter（一对一）', '宽依赖: reduceByKey/join（多对多）'], [False, True]):
    ax.set_xlim(0, 10); ax.set_ylim(0, 10); ax.axis('off')
    ax.set_title(title, fontsize=12, color=('#C00000' if wide else '#548235'))
    for i in range(3):
        ax.add_patch(plt.Rectangle((1, 7.2 - i * 2.4), 2.2, 1.6, fc='#D9E2F3', ec='#2F5597'))
        ax.add_patch(plt.Rectangle((6.8, 7.2 - i * 2.4), 2.2, 1.6, fc='#E2EFDA', ec='#548235'))
        ax.text(2.1, 8 - i * 2.4, f'父P{i}', ha='center', va='center', fontsize=9)
        ax.text(7.9, 8 - i * 2.4, f'子P{i}', ha='center', va='center', fontsize=9)
    if wide:
        for i in range(3):
            for j in range(3):
                ax.plot([3.2, 6.8], [8 - i * 2.4, 8 - j * 2.4], color='#C00000', lw=0.8, alpha=0.5)
    else:
        for i in range(3):
            ax.plot([3.2, 6.8], [8 - i * 2.4, 8 - i * 2.4], color='#548235', lw=1.8)
plt.tight_layout(); plt.show()
Notebook 已保存的运行结果
print('''窄依赖（Narrow）：父分区只被子分区之一使用
  map/filter/flatMap/union -> 无需 Shuffle，可流水线
  宽依赖（Wide）  ：父分区被多个子分区使用
  groupByKey/reduceByKey/join -> 需要 Shuffle（数据跨节点传输）

优化面试点：
  1) reduceByKey 优于 groupByKey（map 端预聚合）；
  2) Shuffle 是性能瓶颈 -> 控制分区数、避免过多 Shuffle；
  3) 血缘长 -> checkpoint 截断（故障恢复更快）。
''')
# 演示：reduceByKey 的 map 端预聚合
partitions = [[('a', 1), ('b', 1)], [('a', 1), ('a', 1)], [('b', 1), ('c', 1)]]
print('分区内预聚合（combine）:')
for i, p in enumerate(partitions):
    d = {}
    for k, v in p:
        d[k] = d.get(k, 0) + v
    print(f'  分区{i}: {p} -> {list(d.items())}')
窄依赖（Narrow）：父分区只被子分区之一使用
  map/filter/flatMap/union -> 无需 Shuffle，可流水线
  宽依赖（Wide）  ：父分区被多个子分区使用
  groupByKey/reduceByKey/join -> 需要 Shuffle（数据跨节点传输）

优化面试点：
  1) reduceByKey 优于 groupByKey（map 端预聚合）；
  2) Shuffle 是性能瓶颈 -> 控制分区数、避免过多 Shuffle；
  3) 血缘长 -> checkpoint 截断（故障恢复更快）。

分区内预聚合（combine）:
  分区0: [('a', 1), ('b', 1)] -> [('a', 1), ('b', 1)]
  分区1: [('a', 1), ('a', 1)] -> [('a', 2)]
  分区2: [('b', 1), ('c', 1)] -> [('b', 1), ('c', 1)]

6️⃣ Cache 与血缘（面试速答）

print('''1) cache():    把中间 RDD 存内存，避免反复从源头重算；
2) checkpoint: 写到可靠存储，切断血缘（适合血缘超长）；
3) 缓存 vs 检查点：缓存可能丢失（依赖血缘可重算），检查点直接切断；
4) 何时用 cache？-> 同一 RDD 被多次 action / 迭代计算；
5) cache 是 lazy 的：真正缓存发生在第一次 action 后。
''')
1) cache():    把中间 RDD 存内存，避免反复从源头重算；
2) checkpoint: 写到可靠存储，切断血缘（适合血缘超长）；
3) 缓存 vs 检查点：缓存可能丢失（依赖血缘可重算），检查点直接切断；
4) 何时用 cache？-> 同一 RDD 被多次 action / 迭代计算；
5) cache 是 lazy 的：真正缓存发生在第一次 action 后。

7️⃣ DataFrame vs RDD（面试对比）

print('''| 维度 | RDD | DataFrame |
|---|---|---|
| 抽象 | 无结构数据+函数 | 有 schema 的表 |
| 优化 | 靠手写 | Catalyst 优化器自动 |
| 类型安全 | 弱（运行时） | 有（编译期 schema） |
| 易用性 | 低 | 高（SQL 语法） |
| 现状 | 底层引擎 | 生产主力（Spark SQL） |

Spark SQL 三大优势：Catalyst 优化、Tungsten 内存管理、DataSource API。
''')
| 维度 | RDD | DataFrame |
|---|---|---|
| 抽象 | 无结构数据+函数 | 有 schema 的表 |
| 优化 | 靠手写 | Catalyst 优化器自动 |
| 类型安全 | 弱（运行时） | 有（编译期 schema） |
| 易用性 | 低 | 高（SQL 语法） |
| 现状 | 底层引擎 | 生产主力（Spark SQL） |

Spark SQL 三大优势：Catalyst 优化、Tungsten 内存管理、DataSource API。

执行流程与分区数设置（面试展开）

**spark-submit 之后发生了什么**：
Driver 启动 → 解析代码构建 DAG → 遇到 **Action** 触发 **Job** → DAGScheduler 按**宽依赖**切 Stage
→ 每个 Stage 生成一批 **Task**（1 分区 = 1 Task）→ TaskScheduler 把 Task 派发给 **Executor** 执行
→ 结果回传 Driver。

**分区数怎么定**（经验法则）：

• 下限：≥ 总核心数（吃满并行度）；

• 经验值：**总核心数的 2~3 倍**（让任务不均时忙核可接活，避免长尾）；

• 太少：并行不足、单任务内存压力大；太多：调度开销、小文件（写出的文件数≈分区数）。

**Shuffle 分区数**由 `spark.sql.shuffle.partitions`（默认 200）控制，小数据调小、大数据调大。

📝 自测清单

• [ ] 能解释 RDD/Transformation/Action/DAG 四概念

• [ ] 能演示惰性求值（action 触发）

• [ ] 能讲清宽窄依赖与 Shuffle

• [ ] 能解释血缘与 cache/checkpoint 的区别

• [ ] 能对比 RDD 与 DataFrame