# 大数据与数仓 · 86 道高频面试题详解

> 配套《高频面试题.md》使用：先自己想 30 秒，再对照【速答】，面试前重点背【速答】、理解【展开】。
> 标注【代码】的题目有可运行案例（均已在实验环境中验证通过），标注配图的题目在 `assets/` 下有对应示意图。
> 版本口径：Hadoop 2.x/3.x、Spark 3.x、Flink 1.18+/2.0、Kafka 3.x/4.0。

---

## 一、体系与 HDFS 题（1–12）

### 1. Hadoop 三件套是什么？各干什么？

**速答**：HDFS 负责分布式存储（分块 + 多副本），YARN 负责资源管理与作业调度（ResourceManager 申请资源、NodeManager 在各节点执行），MapReduce 是分布式计算框架（Map/Shuffle/Reduce 三阶段）。

**展开**：三件套体现"存算分离"的雏形——存储层（HDFS）与计算层（MR）独立部署，同一个 HDFS 上可以跑 MR、Spark、Flink 多种引擎。广义的 Hadoop 生态还包括 Hive（SQL 化）、HBase（KV 存储）、ZooKeeper（协调）、Sqoop/DataX（同步）等。

### 2. HDFS 为什么块要设大（128MB/256MB）？

**速答**：为了把"寻址开销"摊薄。磁盘一次寻址约 10ms，顺序传输约 100MB/s；块设到 128MB 后，寻址时间占比降到千分之一量级，读写接近纯顺序吞吐。同时块越大，NameNode 需要管理的元数据条目越少（1GB 文件只需 8 条块记录）。

**展开**：块从 1.x 的 64MB 升到 2.x 的 128MB（`dfs.blocksize`），就是这个逻辑的延续。但块不能无限大：块太大 → 单任务处理时间长、失败重跑代价高、split 难以切出足够并行度。**块是存储/容错单位，不是读取最小单位**——读取仍按字节流，客户端可以只读块内一段。

### 3. 3 副本策略？副本放置规则（机架感知）？

**速答**：默认 3 副本，放置规则：第 1 副本在客户端所在节点（集群外则随机选一个），第 2 副本放另一个机架的某节点，第 3 副本与第 2 副本同机架的不同节点。

**展开**：这样兼顾两件事——**容错**：任一节点坏有同机架副本，任一机架断电仍有 1 个跨机架副本；**带宽**：写路径只在第 2 副本时跨一次机架交换机，第 3 副本走同机架内网。读的时候客户端优先取最近的副本（同节点 → 同机架 → 跨机架）。Hadoop 3.x 对冷数据可改用**纠删码 RS(6,3)**，存储开销从 3 倍降到 1.5 倍，代价是读放大与修复成本。

配图：`assets/fig_hdfs_blocks.png`

### 4. NameNode 与 DataNode 分工？NameNode 挂了怎么办（HA）？

**速答**：NameNode 管元数据（目录树、文件→块映射，常驻内存，靠 fsimage + editlog 持久化）；DataNode 存实际块数据并周期心跳、汇报块位置。NameNode 单点用 HA 解决：Active/Standby 两台 NameNode，editlog 写入共享的 JournalNode 集群（Quorum），ZKFC 基于 ZooKeeper 做选主与故障切换。

**展开**：两个易错点：①块到 DataNode 的位置信息**不持久化**，重启后由 DataNode 汇报重建（避免元数据与实际数据不一致）；②内存量级——每块元数据约 150 字节，10 亿块就要数百 GB 内存，所以超大集群用 Federation 拆多个命名空间。

### 5. 读写流程？为什么数据流不过 NameNode？

**速答**：写：Client 向 NameNode 申请 → NameNode 返回 DataNode 列表 → Client 把块写入第一个 DataNode，DataNode 间以流水线复制副本 → 全部确认后提交。读：Client 向 NameNode 取"块→节点列表" → 就近并行从 DataNode 拉块。数据流不过 NameNode，因为 NameNode 是单点元数据服务，若再承载全部数据带宽会立刻成为瓶颈——它只回答"去哪拿"，不搬运"拿什么"。

配图：`assets/fig_hdfs_rw.png`

### 6. 小文件问题是什么？怎么解决？

**速答**：每个文件、每个块都要占 NameNode 一条内存元数据（约 150 字节），1 亿个 1KB 小文件仅元数据就要几十 GB 内存，且读写时寻址与 RPC 次数暴涨、Map 任务数=文件数导致大量碎片任务。解决：①源头合并（采集端攒批再落 HDFS）；②定期 Compaction（按分区合并成大 Parquet/ORC）；③HAR/SequenceFile 归档；④改用湖格式（Iceberg/Hudi 自带小文件合并）；⑤Map 端用 CombineFileInputFormat 把多小文件合并成一个 split。

### 7. 存算分离是什么？好处与代价？

**速答**：存储（对象存储 S3/OSS 或独立 HDFS）与计算（Spark/Flink/Presto 集群）物理分开、独立扩缩容。好处：弹性（白天扩计算跑报表、晚上缩掉）、成本（冷数据放廉价对象存储）、多引擎共享同一份湖数据。代价：数据本地性消失，计算要跨网络拉数据 → 依赖高速网络与缓存（Alluxio/本地盘缓存）弥补。

**展开**：传统 Hadoop 是存算耦合（DataNode 同时是计算节点，本地性优先）；云原生架构几乎都是存算分离，这是两种架构最本质的差别。

### 8. HDFS 适合什么场景？不适合什么？

**速答**：适合：大文件、高吞吐、顺序读、一次写多次读（日志、数仓底座、模型数据）。不适合：①海量小文件（元数据爆炸）；②低延迟随机点查（毫秒级 KV 查询请用 HBase/Redis）；③频繁修改（HDFS 只支持追加，不支持随机改）；④超低延迟交互式扫描（走 Doris/ClickHouse）。

### 9. 数据湖 vs 数仓的区别？Lakehouse（湖仓一体）是什么？

**速答**：数仓：先建模后入库（schema-on-write），数据治理好、BI 查询快，但上数据流程重、存结构化数据；数据湖：先存原始数据后建模（schema-on-read），存日志/文本/图像等任意格式、成本低、灵活，但缺事务与治理，容易变"数据沼泽"。Lakehouse：在湖存储（S3/HDFS + Parquet/ORC 开放格式）之上加一层表格式（Iceberg/Hudi/Delta），补上 ACID、元数据、快照、upsert，做到"湖的成本 + 仓的管理"。

### 10. Iceberg / Hudi / Delta Lake 解决什么问题？

**速答**：三者都是**开放表格式（table format）**，解决数据湖上的四大问题：①ACID 事务（并发读写隔离）；②快照与时间旅行（按历史版本查询/回滚）；③行级更新（upsert/delete，流式入库）；④schema 演化与小文件治理。

**展开**：Iceberg 用独立的元数据快照树，引擎中立（Spark/Flink/Trino/StarRocks 都能读），隐藏分区让"按天分区"对用户透明；Hudi 主打 upsert（COW 写时复制 / MOR 读时合并两种表），CDC 入库场景常见；Delta 与 Spark/Databricks 生态结合紧密。选型口径：多引擎共存选 Iceberg，强 upsert/CDC 选 Hudi，All-in Spark 选 Delta。

### 11. 冷热数据分层存储怎么做？

**速答**：按访问频率把数据放到不同介质：热数据（近 7–30 天，高频查询）放 SSD/内存缓存；温数据放 HDD；冷数据（超过半年）转对象存储归档或纠删码。HDFS 原生支持 storage policy（HOT/WARM/COLD/ARCHIVE），Iceberg 可按分区做冷热迁移，Kafka 4.0 起内置分层存储（热数据本地盘、冷数据自动转对象存储）。

### 12. 为什么 HDFS 不适合直接支撑高频点查？

**速答**：三个原因：①随机 IO——点查需要随机寻道，HDFS 设计目标是顺序吞吐；②元数据无索引——NameNode 只有目录树，没有二级索引，定位小对象要完整路径解析 + RPC；③延迟叠加——每次读都要问 NameNode 再连 DataNode，毫秒级延迟对点查场景太慢。正确做法：KV 化（HBase 用 LSM 树 + rowkey 索引、Redis 内存查询），或数仓列存 + 谓词下推。

---

## 二、MapReduce 题（13–22）

### 13. MR 三个阶段？Map 和 Reduce 各做什么？

**速答**：Map：对每个输入分片的每条记录调用 map()，输出 (k2, v2) 中间结果，任务间完全并行；Shuffle：按 k2 分区（hash%R）、排序、分组，把相同 key 从各 Map 节点传给同一个 Reduce；Reduce：对每个 key 的 value 列表调用 reduce() 聚合，输出结果到 HDFS。

**展开**：词频统计：`map(行号, 行文本) → [(词,1)...]`；`reduce(词, [1,1,...]) → sum`。Map 任务数 = 分片数（默认一片=一个 HDFS 块，保证本地读取），Reduce 任务数由 `job.setNumReduceTasks(R)` 指定。

配图：`assets/fig_mr_flow.png`

### 14. Shuffle 是什么？为什么慢？

**速答**：Shuffle 是"Map 输出到 Reduce 输入之间"的全过程：**分区、排序、溢写落盘、归并、网络传输、Reduce 端归并**。慢在三次代价叠加：①磁盘 IO——Map 输出写入环形缓冲区（默认 100MB），到 80% 溢写临时文件，最后归并成一个已排序文件；②排序——溢写前排序、归并时再排序，比较开销大；③网络——Reduce 通过 HTTP 从每个 Map 节点拉取自己的分区。

**展开**：优化三板斧：Combiner（map 端预聚合减少传输量）、压缩中间数据（Snappy/LZ4，CPU 换带宽）、调参（环形缓冲 `mapreduce.task.io.sort.mb`、溢写阈值 0.8、reduce 拉取线程数）。

### 15. Combiner 的作用？

**速答**：Combiner 是可选的"本地 mini-reduce"，在 Map 节点内先聚一次再传输。对求和/计数/最大最小值这类满足结合律的运算，能成倍减少 Shuffle 传输量。注意：Combiner 可能被调用 0 次或多次，逻辑必须与调用次数无关；**求平均不能直接 Combiner**——先平均再平均不等于总平均，要传 (sum, count) 二元组。

【代码】（实验 01 已验证）：

```python
records = [('一班', 85), ('一班', 78), ('一班', 90), ('二班', 92), ('二班', 95)]
acc = {}
for cls, sc in records:
    s, c = acc.get(cls, (0, 0))
    acc[cls] = (s + sc, c + 1)          # Combiner 输出 (sum, count)
avg = {cls: round(s / c, 2) for cls, (s, c) in acc.items()}
# avg['一班'] = 84.33 正确；若 Combiner 输出局部均值 81.5，再与 90 平均得 85.75，错误
```

### 16. 分区器（Partitioner）干什么？

**速答**：决定每条 Map 输出去哪个 Reduce 任务，默认 HashPartitioner：`partition = hash(key) % numReduceTasks`。相同 key 必进同一分区——这是"按 key 分组"跨机器成立的根本保证。可自定义：如按 key 范围分区实现全局有序输出（配合 TotalOrderPartitioner + 采样），或把倾斜的大 key 单独定向。

### 17. MR 为什么被 Spark 替代？

**速答**：①中间结果必须落盘，而 Spark 窄依赖流水线在内存中直通，迭代类任务快 10–100 倍；②MR 只支持 Map→Reduce 两层模型，Spark 有完整 DAG；③MR 每个作业 JVM 冷启动，Spark Executor 常驻复用；④生态：Spark SQL/Structured Streaming/MLlib 一栈解决。

**展开**：MR 没有消失——一些超大规模"单遍聚合"和历史 Hive 集群仍在用，且 Hive 3/4 默认引擎已转向 Tez。

### 18. 数据量超大时怎么避免 OOM？

**速答**：分三处看。Map 端：逐条流式处理天然不 OOM，注意别在 map 里攒全量容器。Reduce 端：①reduce() 的 values 是**迭代器**（边拉边用），不要 `for` 一次存进大 List；②大 key（一个 key 上亿 value）要拆分或改写逻辑。Join 大表：分桶 join（两表按 join key 分桶排序）或改写为 map-side join（小表广播）。框架本身靠**外排**（溢写+归并）避免排序 OOM——这也是它慢的原因之一。

### 19. MR 处理 TopN 的思路？

**速答**：分治：每个 Map 任务只维护**局部 TopN**（TreeMap，内存 O(N)），输出 N 条到 Reduce；Reduce 归并所有局部 TopN 得全局 TopN。数据量从"全部记录"降到"分区数 × N"。

【代码】（实验 07 已验证，配图 `assets/fig_topn.png`）：

```python
def topn(nums, n):
    return sorted(nums, reverse=True)[:n]

partitions = [[5, 9, 3], [10, 2, 7], [8, 1, 6]]
local = [topn(p, 3) for p in partitions]              # 每分区局部 Top3
global_top = topn([x for p in local for x in p], 3)   # 归并
assert global_top == [10, 9, 8]
```

### 20. 什么是数据本地性？

**速答**：把计算任务调度到**数据所在节点**执行，让磁盘读走本地盘而非网络。Hadoop/YARN 调度优先级：同节点 > 同机架 > 跨机架，等待本地槽位有超时（默认几秒），超时降级到机架/远端（避免干等）。这就是"移动计算比移动数据便宜"的落地。存算分离架构下数据本地性减弱，改靠缓存弥补。

### 21. 一个 Reduce 和多个 Reduce 的取舍？

**速答**：单 Reduce：无并行、无归并、天然全局有序，但吞吐受限于单节点——适合数据量小或必须全局序（如全局 TopN 收尾、全局排名）；多 Reduce：并行吞吐高，但只有分区内有序、输出 R 个文件，倾斜时最慢的 task 决定整体时长。工程经验：先估算各 key 分布，热点场景必须配合倾斜治理。

### 22. MR 能处理实时数据吗？为什么不行？

**速答**：不能。MR 是批模型：①输入必须是有界静态数据集；②中间落盘 + 作业冷启动，延迟秒到分钟级；③没有状态管理与事件时间语义。实时链路选 Flink（逐条、毫秒级、有状态）或 Spark Streaming（微批、秒级）。

---

## 三、Spark 题（23–34）

### 23. RDD 是什么？为什么"弹性"？

**速答**：RDD（Resilient Distributed Dataset）是不可变的、分区的、可并行操作的分布式数据集。"弹性"体现在四点：①分区可自动并行；②故障时沿**血缘**重算恢复（不需要数据副本）；③数据可在内存与磁盘间自动降级（内存不够落盘）；④分区数与存储位置可动态调整。

**展开**：RDD 内部记录五元信息：分区列表、依赖（血缘）列表、计算函数（对每个分区怎么算）、分区器（可选，Key 类型 RDD）、优先位置列表（本地性提示）。

### 24. Transformation 与 Action 区别？惰性求值的好处？

**速答**：Transformation（map/filter/reduceByKey 等）只记录转换、不立即执行；Action（collect/count/reduce/save 等）触发真正的作业。惰性的好处：①框架看到完整 DAG 后做全局优化（合并相邻窄转换、谓词下推、裁剪未使用分支）；②不为中间结果付落盘/网络成本；③血缘认证清晰，代码"声明式"，出错可重放。

【代码】（实验 02 已验证）：

```python
class MiniSpark:
    def __init__(self, partitions):
        self._parts = partitions
        self._plan = []          # 只记录转换，不执行
        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 run(self):               # Action：此刻才逐分区流水线执行
        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]
            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)
assert ms._executed == 0          # 构建阶段：0 次遍历
res = ms.run()
assert ms._executed == 2 and sum(len(p) for p in res) == 7
```

### 25. DAG 是什么？Action 时怎么执行？

**速答**：DAG 是"从源 RDD 到目标 RDD"的转换依赖图。Action 触发 Job 后：DAGScheduler **从后往前**遍历 DAG，遇到**宽依赖就切一个 Stage**（Shuffle 边界），窄依赖留在同一 Stage 内流水线执行；每个 Stage 生成"分区数 = Task 数"的 TaskSet 交给 TaskScheduler 派发。

配图：`assets/fig_spark_lazy.png`

### 26. 血缘（Lineage）有什么用？cache 和 checkpoint 区别？

**速答**：血缘记录"每个 RDD 由哪些父 RDD 经什么转换得到"，用途是**容错**：某个分区丢失，只重算丢失分区对应的窄依赖链，不必全量重跑。cache：把 RDD 快照放内存/本地盘，**血缘保留**，快照丢了还能重算，且是 lazy 的（首次 Action 才真正缓存）；checkpoint：把 RDD 写到**可靠存储**（HDFS），并**切断血缘**，适合血缘超长、迭代算法（如 PageRank 上百轮）的场景。用前记得 `sc.setCheckpointDir()`。

### 27. 宽依赖 vs 窄依赖？为什么宽依赖要 Shuffle？

**速答**：窄依赖：每个父分区最多被一个子分区消费（map、filter、union、mapPartitions）——同 Stage 流水线、无 Shuffle、容错可精确重算；宽依赖：父分区被多个子分区消费（reduceByKey、groupByKey、join 非同分区、repartition）——数据必须**按 key 重新分布**到目标分区，跨节点网络传输即 Shuffle。

配图：`assets/fig_spark_deps.png`

### 28. reduceByKey 为什么优于 groupByKey？

**速答**：reduceByKey 在每个分区内部**先做 map 端预聚合（combine）**，跨网络传输的是"每个分区内每个 key 的局部和"；groupByKey 把每个 key 的**全部原始值**都传到 Reduce 端再聚合——Shuffle 量可能大一个数量级，且 Reduce 端要为 value 列表在内存中缓冲，大 key 直接 OOM 风险。

【代码】（实验 02 已验证）：

```python
partitions = [[('a', 1), ('b', 1)], [('a', 1), ('a', 1)], [('b', 1), ('c', 1)]]
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())}')
# 传输量对比：groupByKey 传 6 条原始记录；reduceByKey 只传 4 条 (局部和)
```

### 29. DataFrame vs RDD？Catalyst 优化器做什么？

**速答**：RDD 是"无结构 + 函数"的黑盒；DataFrame 是带 schema 的分布式表，SQL/DSL 表达。DataFrame 的优势在 **Catalyst 优化器**：解析出逻辑计划 → 规则优化（谓词下推、列裁剪、常量折叠、join 重排）→ 生成物理计划（多实现择优）→ **Tungsten** 做全阶段代码生成与堆外二进制内存管理。RDD 只在需要精细控制分区与自定义数据分布时使用。

**展开**：生产代码原则：能 SQL/DataFrame 就不用 RDD；UDF 性能差（序列化开销），优先用内置函数。

### 30. 什么是数据倾斜？怎么发现？怎么解决？

**速答**：key 分布不均导致个别 task 的数据量/耗时远超其他 task，整个 Stage 被长尾拖住。**发现**：Spark UI 看 task 耗时与 shuffle read bytes 的长尾（如 199 个 task 各 1s、1 个 task 10min）；SQL 层面先看 key 的频次分布。**解决**：①两阶段聚合——key 加随机前缀打散局部聚合，再去前缀全局聚合（仅限聚合类）；②热点 key 拆分——把特大 key 单独拎出来单独处理或广播；③join 倾斜——小表广播（broadcast hash join）绕过 shuffle；④过滤/修正异常 key（null、测试账号）；⑤提高 shuffle 分区数（缓解但不根治）。

【代码】（实验 07 已验证，配图 `assets/fig_skew.png`）：

```python
import random
records = [('hot', 1)] * 100 + [('a', 1)] * 10 + [('b', 1)] * 5

def two_phase_agg(records, n_rand=3):
    phase1 = {}
    for k, v in records:
        p = f'{k}#{random.randint(0, n_rand - 1)}'   # 第一阶段：加随机前缀打散
        phase1[p] = phase1.get(p, 0) + v
    phase2 = {}
    for pk, v in phase1.items():
        k = pk.split('#')[0]                        # 第二阶段：去前缀合并
        phase2[k] = phase2.get(k, 0) + v
    return phase2

res = two_phase_agg(records)
assert res['hot'] == 100 and res['a'] == 10 and res['b'] == 5
```

### 31. 分区数怎么设置？

**速答**：下限不低于总核心数（吃满并行度），经验值取**总核心数的 2–3 倍**——任务耗时不均时，忙核可以接长尾任务。读 HDFS 时分区数≈块数较自然；Shuffle 分区由 `spark.sql.shuffle.partitions`（默认 200）控制。分区过多：调度开销大、每个 task 有效工作太少、写出的文件数≈分区数（小文件）；分区过少：并行度不足、单 task 内存压力大、倾斜放大。

### 32. Spark 执行流程？

**速答**：`spark-submit` → 启动 Driver（解析代码建 DAG，不执行）→ 首个 Action 触发 Job → DAGScheduler 按宽依赖切 Stage → 各 Stage 生成 TaskSet → TaskScheduler 把 Task 派发给 Executor（考虑数据本地性）→ Executor 用线程池执行 Task → ShuffleMapTask 写中间数据、ResultTask 产最终结果 → 结果回传 Driver。容错：task 失败重试（默认 4 次），stage 失败重试，长尾可用**推测执行**（对慢 task 冗余起备份）。

### 33. 广播变量与累加器？什么时候用广播？

**速答**：广播变量：把 Driver 端的只读变量（如维表字典、规则映射）以高效方式发到每个 Executor **一份**（否则每个 task 序列化传输一份，大对象开销巨大）。典型场景：**大表 join 小表**——把小表广播，走 broadcast hash join，完全避免 Shuffle。累加器：只能"加"的分布式变量（如统计坏行数），Executor 端更新、Driver 端读取；注意 transformation 里更新累加器在重算时会重复累加，只在 action 或幂等场景用。

### 34. Spark SQL 与 Hive 的关系？

**速答**：Spark SQL 可以直接读取 Hive Metastore 管理的表（`enableHiveSupport()`），兼容大部分 HiveQL 语法与 UDF，做到"表不动、换引擎"——同一份数据，执行引擎从 MR/Tez 换成 Spark（Catalyst + Tungsten），通常快数倍。差异点：个别函数与类型转换行为不同、事务表/ACID 支持有限（Hive ACID 表在 Spark 里只读受限）。数仓演进口径：**Metastore 统一管元数据，计算引擎按需混用**。

---

## 四、Flink 与实时计算题（35–46）

### 35. 批处理 vs 流处理区别？

**速答**：批：数据有界、一次性拉起作业、延迟高（分钟/小时）、结果确定性天然保证；流：数据无界、作业常驻、延迟低（毫秒/秒）、必须处理乱序与状态恢复。本质差别是"数据是否有终点"——Flink 用同一套引擎跑批（有界流）与流（无界流）实现流批一体。

### 36. Lambda vs Kappa 架构？

**速答**：Lambda：批链路（准）+ 流链路（快）并行计算，查询层合并结果。问题：同一逻辑两套代码（Java+SQL/Hive+Spark+Flink），口径漂移、双倍维护成本、对账繁琐。Kappa：一切皆流，需要历史结果时**重放消息**从头算。前提：Kafka 保留期够长（或能重置到任意 offset）、重放吞吐扛得住。现状：多数公司是"Lambda 简化版"——实时链路为主，离线链路做对账与重刷兜底。

### 37. 事件时间 vs 处理时间 vs 摄入时间？

**速答**：事件时间：数据**发生**的时刻（业务时间戳字段），正确性标准；处理时间：算子机器时钟，低延迟但不可复现（重启/反压都会扭曲）；摄入时间：进入 Flink 的时刻，介于两者之间，少用。严肃统计（计费、报表）必须事件时间 + Watermark；监控大盘秒级刷新可接受处理时间。

### 38. 乱序数据怎么处理？Watermark 是什么？

**速答**：Watermark 是随流流动的"时间进度条"，定义为 `wm = max(已见事件时间) - 允许乱序容忍 d`，单调递增。含义：**事件时间 ≤ wm 的数据基本到齐**。窗口关闭条件：wm 推进到窗口终点。d 是准确性与延迟的折中：d 太小关窗早→漏数据；太大关窗晚→结果延迟高。多并行度下，下游取各输入通道 watermark 的**最小值**（木桶效应）。

【代码】（实验 03 已验证，配图 `assets/fig_watermark.png`）：

```python
arrive = [1, 4, 2, 7, 3]      # 按到达顺序的事件时间
allow = 2
wm, late = 0, []
for t in arrive:
    late.append(t < wm)        # 事件时间已落后水位线 -> 迟到
    wm = max(wm, t - allow)    # 水位线单调推进
# 第5条 t=3 到达时 wm=5，3 < 5 -> 迟到，走侧输出/丢弃
```

### 39. Watermark 怎么生成？

**速答**：两种触发：**周期性**（默认每 200ms，`autoWatermarkInterval`）按当前最大事件时间减容忍度生成——生产默认；**逐条/打点**（每条记录后立即生成）。常用策略 `BoundedOutOfOrderness(maxDelay)`（固定乱序容忍）与单调递增（maxDelay=0）。注意**空闲源问题**：某分区长时间无数据会把整体 watermark 卡住，需 `withIdleness` 把空闲源剔除。

### 40. 滚动/滑动/会话窗口区别？

**速答**：滚动窗口：固定长度、不重叠，每条数据只属一个窗口——固定周期报表（每 5 分钟 PV）；滑动窗口：长 > 步长、有重叠，一条数据可属多个窗口——"最近 1 小时、每 5 分钟更新"的趋势监控；会话窗口：按**活动间隔 gap** 动态切分，无固定长度，活跃数据顺延窗口——用户行为会话分析。

【代码】（实验 03 已验证，配图 `assets/fig_windows.png`）：

```python
def tumbling_window(events, win_len):
    buckets = {}
    for eid, ts in events:
        w = (ts // win_len) * win_len            # 窗口起点 = ts - ts % win_len
        buckets.setdefault(w, []).append((eid, ts))
    return buckets

def sliding_window(events, win_len, slide):
    buckets = {}
    for eid, ts in events:
        start = 0
        while start <= ts:
            if start <= ts < start + win_len:    # 一条数据可命中多个窗口
                buckets.setdefault(start, []).append((eid, ts))
            start += slide
    return buckets
```

### 41. 迟到数据怎么办？

**速答**：三层策略：①`allowedLateness`：窗口触发后**保留状态**再等一段时间，迟到数据到了就**更新窗口结果**（要求下游支持更新/ retract，如 Kafka 事务、可更新的 OLAP）；②侧输出流（sideOutputLateData）：太迟的数据导出到独立流，人工核对或离线补偿；③丢弃（默认，最简单但最不准）。口径记忆：**Watermark 决定窗口何时关，allowedLateness 决定关了还能不能补**。

### 42. Flink 状态是什么？Checkpoint 怎么做？

**速答**：状态是算子要"记住"的中间结果（窗口累加值、去重集合、最近事件时间）。两类：**Keyed State**（按 key 隔离：ValueState/ListState/MapState/ReducingState）与 **Operator State**（按算子并行子任务隔离，如 Kafka offset）。Checkpoint 基于 **Chandy-Lamport 异步屏障快照**：JobManager 的 CheckpointCoordinator 周期向 source 注入**屏障**；屏障随数据流向下游，算子收齐所有输入通道的屏障后，把自身状态**异步快照**到状态后端（堆内存/RocksDB，大状态用 RocksDB + 增量快照），然后向下游转发屏障；全部算子确认后该 Checkpoint 完成。故障恢复：全图重启，从最近快照恢复状态，source 重放。

### 43. 精确一次（Exactly-once）怎么实现？

**速答**：分两层。**内部精确一次**：Checkpoint 恢复 + 重放，算子状态不多不少。**端到端精确一次**：sink 也要参与——**两阶段提交（2PC）**：屏障到达 sink 时把已算出的结果**预提交**（写入但对外不可见，如 Kafka 事务挂起）；Checkpoint 全部完成后**正式提交**（对外可见）。故障时预提交数据随事务中止丢弃，不会多写。要求 sink 支持事务（Kafka 事务）或幂等写（按主键 upsert）。

### 44. Flink vs Spark Streaming？

**速答**：模型：Flink 真流（逐条处理，毫秒级）；Spark Structured Streaming 微批（默认几百毫秒~秒级攒批，Continuous Mode 仍实验性）。语义：Flink 原生事件时间/Watermark/状态更完善；Spark 起家批处理，流是"增量查询"。生态：Flink 在实时数仓/CEP/风控占优；Spark 在批流混合、与 Hive/湖格式配合时省心。选型口径：强实时低延迟选 Flink，"准实时+重批"一体选 Spark。

### 45. 背压（Backpressure）是什么？怎么处理？

**速答**：下游算子处理不过来，反压逐级向上游传播，最终 source 限速——系统自我保护不丢数据，但延迟飙升。发现：Flink WebUI 的 backpressure 标签（HIGH）、算子 busyTime 高、outputQueueLength 打满、checkpoint 超时（屏障传不动）。处理：①加并行度（受 key 分区数限制的场景先看倾斜）；②优化算子逻辑（避免每条记录重对象创建、减少序列化）；③同步 IO 改异步 IO（维表查询）；④攒批（上游预聚合减少流量）；⑤扩大网络缓冲、调 checkpoint 间隔。

### 46. 实时指标和离线指标对不上怎么办？

**速答**：先按四类原因排查：①**口径**——去重范围（分端去重 vs 全局）、窗口边界（左闭右开）、指标定义不一致；②**时间**——事件时间 vs 处理时间、时区（UTC vs UTC+8）、T-1 分区边界；③**数据**——乱序迟到导致实时少算、上游重放导致实时多算、实时读 Kafka 流离线读全量表数据源不同；④**修复机制**——实时链路是否有 retract/补数能力。治理：建立以离线为准的对账监控（小时级 diff 告警），超阈值自动用离线结果校准/重刷实时。

---

## 五、数仓建模题（47–58）

### 47. 数仓为什么分层？各层职责？

**速答**：四层：ODS 原始层（贴源存储、原样保留、可回溯重放）；DWD 明细层（清洗、标准化、维度退化，最细粒度事实）；DWS 汇总层（按主题/维度组合轻度汇总，复用中间结果）；ADS 应用层（面向报表、大屏、算法特征的结果级指标）。价值：**解耦**（上游变化只影响下一层）、**复用**（DWS 算一次多处用）、**质量**（逐层校验，问题早暴露）、**成本**（中间结果只算一次）。

配图：`assets/fig_wh_layers.png`

### 48. ODS/DWD/DWS/ADS 分别做什么？粒度如何变化？

**速答**：粒度自上而下从"与源一致"到"结果级"：ODS 每行=一条源记录（含脏数据）；DWD 每行=一次业务事实（一笔订单/一次点击）；DWS 每行=一个维度组合的汇总（某天某城市的订单量）；ADS 每行=一个应用指标（某日 GMV Top3）。**跨层去重陷阱**：DWS 按 (天, 城市) `COUNT(DISTINCT user_id)` 后，ADS 再 SUM 会重复计数——跨组去重必须回 DWD 明细层单独算。

### 49. 事实表 vs 维度表？

**速答**：事实表存"业务过程的可加度量"（订单金额、点击次数），行数巨大、粒度明确（一次交易一行）；维度表存"描述上下文的实体属性"（用户、商品、日期），行数少、字段多。事实表通过外键关联维度表。特殊事实表：周期快照（每日库存余额）、累积快照（订单从下单到签收全流程）、无事实事实表（仅记录"某学生某天到课"）。

### 50. 星型模型 vs 雪花模型？怎么选？

**速答**：星型：维度表直连事实表、维度冗余——join 少、查询快、建模简单，BI 报表首选；雪花：维度表再规范化拆分（如商品维拆出品类维）——省存储、一致性更好，但 join 层数多、查询慢。**选型口诀：查询性能优先选星型，存储/一致性优先选雪花**；实践中数仓 DWD/DWS 层普遍星型 + 宽表，规范化留给源业务库。

配图：`assets/fig_star_snow.png`

### 51. 什么是宽表（维度退化）？什么时候用？

**速答**：把常用维度字段直接冗余进事实/汇总表（如 dws_daily_orders 里直接带 city 名、品类名），查询时不再 join 维度表——空间换时间，是 DWS/ADS 层的常规操作，OLAP 引擎（Doris/ClickHouse）的大宽表就是这一思路的极致。代价：维度变更要回刷历史、存储冗余。适合：维度变化少、查询频繁、对延迟敏感的场景。

### 52. 指标口径不统一怎么治理？

**速答**：四步：①**统一命名**——指标字典（原子指标 + 修饰词 + 时间周期，如"日 GMV_有效订单"），同义指标合并；②**统一计算**——口径只在 DWS 层落地一次，下游只能引用不能重算；③**统一来源**——明确每个指标的源表与过滤条件（如"有效订单"排除哪些 status）；④**对账监控**——关键指标设校验任务，数出多门自动告警。组织上要有数据 BP/指标委员会拍板争议口径。

### 53. 数据质量四维度？

**速答**：①**完整性**——行数对不对、字段缺失率（主键唯一、非空）；②**准确性**——值域合理（年龄 0–150）、波动检测（今日 GMV 环比 ±50% 告警）、与源系统对账；③**一致性**——同名指标多表一致（跨表主外键、口径一致）；④**及时性**——任务按时完成（SLA 告警）、数据新鲜度。落地：调度里加质量稽核节点（质量不过→阻断下游 + 告警）。

### 54. 元数据管理包含什么？血缘有什么用？

**速答**：元数据分三类：技术元数据（表结构、分区、存储位置、行数）、业务元数据（指标口径、字段含义、责任人）、操作元数据（任务依赖、运行历史、资源消耗）。**血缘**记录"表/字段的上下游关系"，三大用途：①影响分析（改表前看下游波及面）；②故障溯源（报表异常沿血缘向上定位）；③治理（识别无下游孤儿表下线、识别高频高频链路做优化）。

配图：`assets/fig_lineage.png`

### 55. 缓慢变化维（SCD）Type1/Type2？

**速答**：维度属性变化怎么存。Type 1：直接覆盖（UPDATE）——省事但丢历史，适合"改错字"；Type 2：加版本（旧记录封版 end_date=变更日前一天，新记录开链到 9999-12-31）——历史可溯，适合用户地址、商品品类。变体：Type 3 只存"上一版值"（加一列 previous_value），处理能力有限。拉链表是 Type 2 的工程化落地。

【代码】（实验 04 已验证）：

```python
import sqlite3
conn = sqlite3.connect(':memory:')
cur = conn.cursor()
cur.execute('CREATE TABLE d_user_zip (user_id TEXT, city TEXT, start_date TEXT, end_date TEXT)')
cur.execute("INSERT INTO d_user_zip VALUES ('u1', '北京', '2024-01-01', '9999-12-31')")

def zipper_update(user, new_city, eff_date):
    cur.execute("UPDATE d_user_zip SET end_date = ? WHERE user_id=? AND end_date='9999-12-31'",
                (eff_date, user))                      # 1) 闭链
    cur.execute('INSERT INTO d_user_zip VALUES (?,?,?,?)',
                (user, new_city, eff_date, '9999-12-31'))  # 2) 开链
    conn.commit()

zipper_update('u1', '上海', '2024-04-01')

q = lambda d: cur.execute("SELECT city FROM d_user_zip WHERE user_id='u1' "
                          "AND ? >= start_date AND ? <= end_date", (d, d)).fetchone()[0]
assert q('2024-03-15') == '北京' and q('2024-05-01') == '上海'
print('拉链表：3月15日在北京，5月1日在上海 —— 历史可溯')
```

### 56. 数据倾斜在 SQL/MapReduce 里怎么处理？

**速答**：聚合类：①两阶段聚合——`GROUP BY key, floor(rand()*N)` 局部聚合后再按 key 全局聚合；②热点 key 拆出单独处理。Join 类：③小表广播（map-side join）绕过 shuffle；④大 key 单独拎出广播 join，其余正常 join；⑤倾斜 key 加随机前缀扩容匹配（空间换时间）。前提动作：先 `GROUP BY key` 看 key 分布确认倾斜点，能过滤的异常 key（null/测试数据）先过滤。

### 57. 为什么数仓不建议大量 join？怎么优化？

**速答**：分布式 join 是最贵的算子（Shuffle 网络 + 排序 + 大表扫描），且 join 层数多时优化器难以选择执行序、口径易错。优化：①**维度退化/宽表**——DWD 生成时就把常用维度冗余进去；②**预聚合**——DWS 层把高频分组先算好；③必须 join 时按代价选算法：小表广播 > 分桶 join（两表按 key 预分桶排序，merge join 免 shuffle）> shuffle hash join > sort merge join。

### 58. 拉链表是什么？怎么更新？

**速答**：拉链表 = SCD Type 2 的落地表结构：`(业务主键, 属性, start_date, end_date)`，end_date=9999-12-31 表示当前有效。每日更新两条 SQL：**闭链**（把有变化的当前记录 end_date 改为生效日前一天）+ **开链**（插入新版本）。优点：全量快照表存 N 天 = N 份数据，拉链表只存"变化量"，适合亿级用户/商品维表；查询任意历史时间点用 `start_date <= d <= end_date` 过滤。注意：需要每日比对"今日全量 vs 昨日拉链"找变化（哈希比对全字段）。

---

## 六、Kafka 与管道题（59–70）

### 59. Kafka 在数据链路中起什么作用？

**速答**：四个作用：①**解耦**——上游只管往 Topic 写，下游各自消费，互不影响、可随时增减；②**削峰**——突发流量先入队列，下游按消费能力匀速处理（如大促日志洪峰）；③**缓冲与持久化**——消息落盘可回放，下游故障重启后从 offset 续传不丢；④**分发的总线**——一份日志同时供实时计算、数仓入库、搜索索引多条链路使用。

配图：`assets/fig_kafka.png`

### 60. Topic / Partition / Offset / Broker 概念？

**速答**：Topic：逻辑消息类别（如 order_events）；Partition：Topic 的物理分片——并行与有序性的单位，每分区是一条**只追加、不可改**的日志，消息在分区内有唯一递增编号 offset（从 0 开始）；Broker：Kafka 服务节点，存分区副本，每分区 1 个 Leader（负责读写）+ N 个 Follower（同步）。分区数决定最大并行度：消费者数超过分区数则多出的闲置。

### 61. 分区内有序还是全局有序？要全局有序怎么办？

**速答**：Kafka 只保证**分区内有序**（追加日志天然有序）。全局有序只能：单分区 Topic（牺牲并行与吞吐，需要全局序的小流量场景可用）；或业务层带序列号/时间戳由消费端重排（如按事件时间窗口内重排）。多数业务其实只需要"同一实体有序"（同一订单/用户），按 key 路由即可满足。

### 62. 为什么同 key 路由同分区？

**速答**：生产者默认分区策略 `hash(key) % 分区数`（Kafka 客户端用 murmur2）。同 key 同分区保证两件事：①**分区内有序**——同一实体的操作序列不乱（先支付后退款不能反）；②**状态局部性**——下游按 key 聚合/Join 时同一实体在同一分区处理，无需跨分区合并。注意：分区数扩容会改变 hash(key)%N 的映射（老 key 换分区、临时乱序），所以**分区数尽量一次规划到位**。

### 63. 消费者组是什么？分区与消费者数量关系？

**速答**：Consumer Group 是协作消费单位：**组内互斥**（一个分区同一时刻只分配给组内一个消费者）、**组间广播**（每个组都拿到全量消息——实时链路一个组、数仓入库另一个组）。数量关系：消费者数 ≤ 分区数才有意义，超出的消费者闲置；理想配比是消费者数=分区数（或整除关系）。消费者增减触发 Rebalance 重新分配分区。

配图：`assets/fig_kafka_offset.png`

### 64. Rebalance 是什么？为什么有代价？

**速答**：消费者组成员变化（上线/下线/心跳超时/订阅变更）时，全组重新分配分区的过程。代价：①Rebalance 期间**整组停止消费**（旧协议 stop-the-world，可能分钟级）；②分区重新分配后，消费者从新分区的**已提交 offset** 续传——若 offset 提交不及时会重复消费。优化：调 `max.poll.interval.ms` 防误判离组、用**静态成员**（group.instance.id，重启不触发）、升级 Kafka 4.0 的 KIP-848 增量协作式 Rebalance（Broker 协调、只迁移受影响分区、不停全组）。

### 65. 消息积压怎么排查和处理？

**速答**：先量化：`最新 offset − 已提交 offset`（Lag）看哪个分区/组积压。排查链路：①消费端慢在哪（处理逻辑慢？外部调用慢？反序列化慢？GC？）——看消费 TPS 与单条耗时；②是不是倾斜（某分区流量大——热点 key 单分区独大）；③消费者挂了或 Rebalance 循环。处理：短期——扩分区+扩消费者（注意扩分区改变 key 路由）、批量拉取批量写、跳过非关键消息事后补偿；中期——优化消费逻辑（异步 IO、攒批写库）、热点 key 拆分；长期——容量规划 + Lag 告警阈值。

### 66. acks=0/1/all 区别？怎么选？

**速答**：acks=0：发出去就算成功，不等确认——最快但 leader 挂/网络抖动会丢；acks=1：Leader 写入本地日志即确认——Leader 应答后未同步给 Follower 就挂，新 Leader 上位后这条丢失；acks=all（-1）：**ISR 中所有副本**确认才算成功——最稳，配合 `min.insync.replicas=2`（ISR 至少 2 个）与 3 副本使用，兼顾可用性与不丢。选型：埋点日志可用 1（可容忍少量丢），订单/支付必须 all。**标准组合拳：3 副本 + acks=all + min.insync.replicas=2 + retries>0 + 消费端幂等**。

### 67. 消费语义：at-most-once / at-least-once / exactly-once？

**速答**：at-most-once：先提交 offset 再处理——崩溃时丢消息（适合埋点可容忍）；at-least-once：先处理再提交 offset——崩溃时重复消费（主流选择，配幂等去重）；exactly-once：处理与提交原子化——Kafka 事务（生产端幂等 + 事务跨分区原子写）或消费-计算-生产闭环（Flink 两阶段提交 + Kafka 事务）。去重手段：唯一业务键幂等写（INSERT ON DUPLICATE KEY / 主键覆盖）、Redis 记录已处理 ID、窗口内去重表。

### 68. Kafka 为什么吞吐高？

**速答**：四件事：①**顺序写**——分区是 append-only 日志，磁盘顺序写数百 MB/s（随机写的百倍）；②**零拷贝**——消费者拉数据走 `sendfile`，从页缓存直达网卡缓冲区，不进用户态、省两次拷贝；③**批量+压缩**——producer 攒批（`linger.ms`、`batch.size`）+ lz4/zstd 压缩，网络与磁盘都省；④**依赖 OS 页缓存**——热数据在页缓存命中，读写基本不碰磁盘。另外 Broker 间与消费者端都是稀疏索引定位（offset→position），查找 O(log n)。

### 69. 重复消费怎么去重？

**速答**：重复来源：Rebalance 后从旧 offset 续传、生产端 retry 重发、offset 提交失败。去重三招：①**写端幂等**——业务唯一键（订单号）做主键，`INSERT ... ON DUPLICATE KEY UPDATE` 或 upsert，天然免疫重复；②**处理日志**——Redis/HBase 记录"已处理的 message ID"，处理前查一下（注意过期时间）；③**事务**——Kafka 事务把"消费-处理-生产-提交 offset"绑成原子操作（配合流处理引擎）。工程首选幂等键设计，把去重下沉到存储层。

### 70. Kafka vs RocketMQ？

**速答**：定位与侧重不同：Kafka 吞吐极高、生态最全（流处理、连接器），但消息可靠性语义偏流式场景，延迟队列/定时消息需自己实现；RocketMQ 面向业务消息：天然支持定时/延迟消息、事务消息（半消息）、消息轨迹、按吞吐调优更省心，金融业务常用。顺序性两者都靠分区/队列内有序。选型：大数据管道（日志、数仓、流计算）选 Kafka；业务交易消息（订单、支付）与需要延迟/事务消息的场景选 RocketMQ（或 RabbitMQ 低流量场景）。

---

## 七、平台与调度题（71–78）

### 71. 调度系统做什么？

**速答**：四件事：①**定时触发**——按 crontab/数据时间周期拉起任务；②**依赖编排**——按 DAG 依赖决定执行顺序（拓扑排序），上游没好下游不跑；③**失败处理**——重试（指数退避）、超限告警、阻断下游；④**重跑与补数**——按分区批量回刷历史（backfill），支持优先级与并发控制。再加运行监控（时长/资源/SLA）与权限审计。

### 72. 数据时间 vs 运行时间？

**速答**：运行时间是任务真实执行的墙钟时间；数据时间（业务日期）是这批数据对应的分区（如 ds=20261007）。天级任务普遍 **T-1**：今天凌晨跑昨天的数据——因为昨天的数据到今天零点才完整。排查问题三连问：任务跑了吗？跑到几点？处理的是哪天的分区？举例："10 月 8 日 01:30 跑的任务，运行时间 10-08，数据时间 10-07。"

### 73. 任务失败怎么处理？

**速答**：四步：①**自动重试**——指数退避（1/5/15min），网络抖动类瞬时故障大多自愈；②**告警分级**——超限后按严重度升级（IM 消息 → 短信 → 电话），带上任务名、失败原因、影响范围（血缘下游）；③**依赖阻断**——上游失败，下游一律不调度，防止用残缺数据算出错误指标污染下游；④**恢复与补数**——修好数据后按分区重跑，验证后通知下游。事后复盘要沉淀监控规则（同样问题第二次自动发现）。

### 74. 任务积压怎么追？

**速答**：先看积压原因：资源不够（队列排队）、单任务变慢（倾斜/数据量涨）、依赖链变长。手段按序：①**提并行度**——扩 YARN 队列资源、调大并发 slot；②**拆任务**——大任务按分区/业务拆成并行子任务；③**优先级调度**——核心链路插队，非关键任务让路；④**跳重跑/合并**——历史补数任务只跑关键分区，非关键基线合并；⑤**增量替代全量**——改增量计算减少单次数据量。预防：大促前容量评估 + SLA 演练。

### 75. 血缘怎么采集与使用？

**速答**：采集三种方式：①**解析 SQL/任务脚本**——静态解析 INSERT/CREATE 的表引用（主流，覆盖度高）；②**运行时拦截**——Hook 执行计划/审计日志拿真实依赖（更准）；③**手工登记**——兜底补漏。使用：①影响分析（改表前查下游）；②故障溯源（异常沿血缘上溯定位源头任务）；③治理（孤儿表下线、高成本链路优化、字段级溯源支撑合规审计）。

配图：`assets/fig_lineage.png`

### 76. 权限与安全？

**速答**：四个层次：①**权限模型**——表级/列级授权（行级通过视图或策略引擎实现），默认最小权限、按角色申请审批；②**脱敏**——手机号/身份证在查询时动态脱敏（掩码、哈希），导出需走审批；③**审计**——所有查询记录"谁、何时、查了什么表、多少行"，异常行为（批量拉取敏感列）告警；④**加密与隔离**——传输 TLS、存储加密，生产/测试环境数据隔离。面试口径：说"我们按最小权限 + 列级脱敏 + 全量审计"即可展开。

### 77. Airflow vs DolphinScheduler？

**速答**：Airflow：Python 代码定义 DAG（DAG as code），插件生态最大，适合工程化团队，但权限与多租户较弱；DolphinScheduler：国产、可视化拖拽 + 代码双模式，原生多租户、分布式去中心化架构（无单点），对大数据组件（Hive/Spark/Flink/DataX）支持友好。另有 Dagster/Prefect（数据感知资产编排）、Argo（K8s 原生）。答题思路：先讲自己实际用过的（怎么定义依赖、怎么重跑、怎么告警），再对比选型理由。

### 78. 一条数据从产生到应用的全链路怎么讲？

**速答**（30 秒版）：业务库/日志产生数据 → **采集**：CDC（Binlog）或 Flume/Filebeat 推 Kafka → **实时支线**：Flink 消费 Kafka 做窗口聚合，产出实时指标/风控特征；**离线支线**：Kafka→HDFS/湖，Hive/Spark 完成 ODS→DWD→DWS→ADS 分层加工 → **应用**：BI 报表、数据 API、算法特征回灌业务。全程由调度编排、血缘追踪、质量稽核、权限管控支撑。追问会沿着任意一环深挖（见前 77 题）。

配图：`assets/fig_pipeline.png`

---

## 八、手撕与 SQL 题（79–86）

> 以下代码均为**完整可运行**版本（单机 Python 3.x 验证通过），面试时先讲思路再写代码。

### 79. 手写 MR 词频统计（Map→Shuffle→Reduce）

```python
def mapper(text):
    """Map：每行文本 -> [(词, 1), ...]"""
    return [(w, 1) for w in text.lower().split()]

def shuffle(map_outputs):
    """Shuffle：按键分组（框架还会排序+按 hash 分发到不同 Reduce）"""
    grouped = {}
    for kv_list in map_outputs:
        for k, v in kv_list:
            grouped.setdefault(k, []).append(v)
    return grouped

def reducer(grouped):
    """Reduce：对每个 key 聚合"""
    return {k: sum(vs) for k, vs in grouped.items()}

lines = ['hello world hello big data', 'big data big data hello']
result = reducer(shuffle([mapper(line) for line in lines]))
assert result['hello'] == 3 and result['big'] == 3 and result['data'] == 3
print(result)   # {'hello': 3, 'world': 1, 'big': 3, 'data': 3}
```

配图：`assets/fig_mr_flow.png`、`assets/fig_wc_bar.png`

### 80. 实现拓扑排序（判断 DAG）

```python
from collections import deque

def topo_sort(graph):
    """graph: {task: [依赖的上游...]}，返回拓扑序；有环返回 None"""
    indeg = {t: len(d) for t, d in graph.items()}
    adj = {t: [] for t in graph}
    for t, ds in graph.items():
        for d in ds:
            adj[d].append(t)
    q = deque(t for t, i in indeg.items() if i == 0)
    order = []
    while q:
        cur = q.popleft()
        order.append(cur)
        for nxt in adj[cur]:
            indeg[nxt] -= 1
            if indeg[nxt] == 0:
                q.append(nxt)
    return order if len(order) == len(graph) else None

tasks = {'ods': [], 'dwd': ['ods'], 'dws': ['dwd'], 'ads': ['dws'], 'feature': ['dws']}
assert topo_sort(tasks) == ['ods', 'dwd', 'dws', 'ads', 'feature']
assert topo_sort({'a': ['b'], 'b': ['a']}) is None      # 环 -> 非法依赖
```

配图：`assets/fig_scheduler_dag.png`

### 81. 手写滚动/滑动窗口

```python
def tumbling_window(events, win_len):
    """滚动窗口：固定长、不重叠。events: [(id, ts)]"""
    buckets = {}
    for eid, ts in events:
        buckets.setdefault(ts - ts % win_len, []).append((eid, ts))
    return buckets

def sliding_window(events, win_len, slide):
    """滑动窗口：长 win_len、步长 slide，一条数据可属多个窗口"""
    buckets = {}
    for eid, ts in events:
        start = (ts - (win_len - 1) - (ts - (win_len - 1)) % slide) if ts >= win_len - 1 else 0
        while start <= ts:
            if start <= ts < start + win_len:
                buckets.setdefault(start, []).append((eid, ts))
            start += slide
    return buckets

stream = [('a', 3), ('b', 5), ('c', 8), ('d', 12), ('e', 17)]
assert sum(len(v) for v in tumbling_window(stream, 10).values()) == 5      # 不重叠：总数=数据数
assert sum(len(v) for v in sliding_window(stream, 6, 3).values()) >= 5     # 重叠：总数>=数据数
```

配图：`assets/fig_windows.png`

### 82. 手写两阶段聚合（抗倾斜）

```python
import random

def two_phase_agg(records, n_rand=3):
    """records: [(key, value)]。热点 key 加随机前缀打散 -> 局部聚合 -> 去前缀全局合并"""
    phase1 = {}
    for k, v in records:
        salted = f'{k}#{random.randint(0, n_rand - 1)}'
        phase1[salted] = phase1.get(salted, 0) + v
    phase2 = {}
    for salted, v in phase1.items():
        k = salted.split('#')[0]
        phase2[k] = phase2.get(k, 0) + v
    return phase2

records = [('hot', 1)] * 100 + [('a', 1)] * 10 + [('b', 1)] * 5
res = two_phase_agg(records)
assert res == {'hot': 100, 'a': 10, 'b': 5}
```

配图：`assets/fig_skew.png`

### 83. 手写全局 TopN 分治

```python
def topn_partition(nums, n):
    """每分区局部 TopN：等价于维护一个大小为 N 的最小堆，这里用排序表达"""
    return sorted(nums, reverse=True)[:n]

def global_topn(partitions, n):
    local = [topn_partition(p, n) for p in partitions]     # 各分区只上传 N 条
    return topn_partition([x for p in local for x in p], n)  # 归并

partitions = [[5, 9, 3], [10, 2, 7], [8, 1, 6]]
assert global_topn(partitions, 3) == [10, 9, 8]
# 正确性：全局 TopK 必在"各分区局部 TopK 的并集"内（反证：若 x 不在其分区 TopK 内，
# 则该分区有 K 个元素比 x 大，全局也有 >= K 个比 x 大，x 不可能进全局 TopK）
```

配图：`assets/fig_topn.png`

### 84. SQL：分组 top1（row_number）

```sql
-- 每个城市最近一笔有效订单
SELECT city, order_id, ts
FROM (
  SELECT city, order_id, ts,
         ROW_NUMBER() OVER (PARTITION BY city ORDER BY ts DESC) AS rn
  FROM dwd_orders
  WHERE is_valid = 1
) t
WHERE rn = 1;
```

【可运行验证】（sqlite3，窗口函数需 SQLite 3.25+）：

```python
import sqlite3
conn = sqlite3.connect(':memory:')
c = conn.cursor()
c.executescript('''
CREATE TABLE dwd_orders (city TEXT, order_id TEXT, ts TEXT, is_valid INT);
INSERT INTO dwd_orders VALUES
 ('北京','o1','2024-05-01 10:00',1), ('上海','o2','2024-05-01 11:00',1),
 ('北京','o3','2024-05-02 09:00',0), ('北京','o4','2024-05-03 12:00',1);
''')
rows = c.execute('''
SELECT city, order_id, ts FROM (
  SELECT city, order_id, ts,
         ROW_NUMBER() OVER (PARTITION BY city ORDER BY ts DESC) AS rn
  FROM dwd_orders WHERE is_valid = 1) t
WHERE rn = 1 ORDER BY city''').fetchall()
assert rows == [('上海', 'o2', '2024-05-01 11:00'), ('北京', 'o4', '2024-05-03 12:00')]
print(rows)
```

要点：`ROW_NUMBER`（强制取一条）vs `RANK`（并列同名次，top1 可能多条）；先 WHERE 过滤再开窗，减少扫描量。

### 85. SQL：日活/月活去重、留存

```sql
-- 日活（DAU）
SELECT dt, COUNT(DISTINCT user_id) AS dau FROM user_active GROUP BY dt;

-- 次日留存
SELECT a.dt AS d1,
       COUNT(DISTINCT a.user_id) AS users,
       COUNT(DISTINCT b.user_id) AS retained,
       ROUND(COUNT(DISTINCT b.user_id) / COUNT(DISTINCT a.user_id), 3) AS retention_1d
FROM user_active a
LEFT JOIN user_active b
  ON a.user_id = b.user_id AND b.dt = DATE_ADD(a.dt, 1)   -- Hive 语法
GROUP BY a.dt;
```

【可运行验证】（sqlite3，DATE_ADD 换成日期整数演示）：

```python
import sqlite3
conn = sqlite3.connect(':memory:')
c = conn.cursor()
c.executescript('''
CREATE TABLE user_active (dt INT, user_id TEXT);
INSERT INTO user_active VALUES
 (1,'u1'),(1,'u2'),(1,'u3'),   -- 第1天 3 人
 (2,'u1'),(2,'u2'),            -- 第2天 u1/u2 回访
 (3,'u1');
''')
# 留存：自连接（b.dt = a.dt + 1）
rows = c.execute('''
SELECT a.dt AS d1, COUNT(DISTINCT a.user_id) AS users,
       COUNT(DISTINCT b.user_id) AS retained
FROM user_active a
LEFT JOIN user_active b ON a.user_id = b.user_id AND b.dt = a.dt + 1
GROUP BY a.dt ORDER BY a.dt''').fetchall()
assert rows == [(1, 3, 2), (2, 2, 1), (3, 1, 0)]    # 第1天留存 2/3，第2天 1/2
print(rows)
```

要点：①活跃口径先说清（"当日至少一次有效行为"）；②大表 JOIN 大表先按天分区裁剪，`COUNT(DISTINCT)` 大时改两阶段（先 `GROUP BY dt, user_id` 再 count）；③留存也可用**自关联窗口**或按天拉平的宽表预计算。

### 86. 手写模拟 HDFS 分块与 3 副本分配

```python
import math, random

def split_file(size_mb, block_mb=128):
    """按块大小切分文件：n = ceil(S/B)"""
    n = math.ceil(size_mb / block_mb)
    sizes = [block_mb] * (n - 1) + [size_mb - block_mb * (n - 1)]
    return sizes

def assign_replicas(n_blocks, n_nodes, racks, replication=3):
    """机架感知简化版：副本必须分布在不同机架；返回 {block: [node...]}"""
    assign = {}
    for b in range(n_blocks):
        chosen, used_racks = [], set()
        while len(chosen) < replication:
            node = random.randrange(n_nodes)
            if node in chosen or racks[node] in used_racks:
                continue                      # 同节点/同机架跳过
            chosen.append(node); used_racks.add(racks[node])
        assign[b] = sorted(chosen)
    return assign

sizes = split_file(500)                       # [128, 128, 128, 116]
assert sum(sizes) == 500 and len(sizes) == 4
racks = ['A', 'A', 'B', 'B', 'C']             # 5 节点 3 机架
assign = assign_replicas(len(sizes), 5, racks)
assert all(len(nodes) == 3 and len({racks[n] for n in nodes}) == 3
           for nodes in assign.values())      # 每块 3 副本且跨 3 机架
print(assign)
```

配图：`assets/fig_hdfs_blocks.png`

---

## 附：90 秒背诵骨架（对照原清单版）

**全链路**：日志/库 → 采集(Kafka) → 存储(HDFS/湖) → 离线(Hive/Spark) → 数仓(ODS→DWD→DWS→ADS) → 报表/特征；实时：→ Flink → 实时指标；调度+血缘+质量贯穿。
**HDFS**：块 128MB、3 副本（机架感知：本节点→另一机架→同机架异节点）、NameNode 元数据/DataNode 数据、小文件是坑、随机点查找 HBase。
**MR**：Map(局部) → Shuffle(分区+排序+落盘+网络) → Reduce(聚合)；Combiner 传 (sum,count) 而非均值；Shuffle 慢、被 Spark 替代。
**Spark**：RDD+惰性求值+血缘；窄依赖流水线、宽依赖 Shuffle 切 Stage；reduceByKey 预聚合；倾斜用两阶段聚合/广播 join；DataFrame+Catalyst 为主。
**Flink**：事件时间+Watermark(max−d) 处理乱序；滚动/滑动/会话窗口；迟到用 allowedLateness/侧输出；精确一次=Checkpoint+两阶段提交；背压查 WebUI、加并行度/异步 IO。
**数仓**：四层解耦复用；事实表可加、维度表描述；星型快/雪花省；宽表最快；口径统一靠指标字典；质量四维（完整/准确/一致/及时）；拉链表=SCD Type2；倾斜打散。
**Kafka**：分区并行、同 key 同分区有序；消费者组组内互斥组间广播；acks=all+min.insync.replicas=2 最稳；高吞吐=顺序写+零拷贝+批量压缩；积压=扩分区扩消费者+优化消费逻辑。
**平台**：DAG+拓扑排序+重试阻断；数据时间 T-1；血缘=影响分析/故障溯源/治理；权限=最小权限+列级脱敏+审计。